switchMap changed to flatMap, Observable resubscribed on an Error (#29)

* switchMap changed to flatMap, Observable resubscribed on an Error * version bumped up
mojaloop · Apr 11, 2019 · 3b6d5f0 · 3b6d5f0
1 parent f9e4628
commit 3b6d5f0
Show file tree

Hide file tree

Showing 5 changed files with 198 additions and 62 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "central-event-processor",
-  "version": "5.3.0",
+  "version": "5.5.0",
   "description": "CEP for Mojaloop Central-Ledger to monitor the notificaion kafka topic and act on it.",
   "main": "app.js",
   "dependencies": {

diff --git a/settlementSubmitter.js b/settlementSubmitter.js
@@ -1,37 +1,140 @@
 const Utility = require('./src/lib/utility')
 
-let message = {
-  'id': '86ce97cf-2302-4d64-8b0e-df5ae9355dd0',
-  'from': 'central-switch',
-  'to': 'dfsp2',
-  'content': {
-    'headers': {
-      'Content-Type': 'application/json',
-      'Date': '2019-01-17T14:46:48.918Z',
-      'FSPIOP-Source': 'central-switch',
-      'FSPIOP-Destination': 'dfsp2'
+let messages = [
+  {
+    'id': '86ce97cf-2302-4d64-8b0e-df5ae9355dd0',
+    'from': 'central-switch',
+    'to': 'dfsp2',
+    'content': {
+      'headers': {
+        'Content-Type': 'application/json',
+        'Date': '2019-01-17T14:46:48.918Z',
+        'FSPIOP-Source': 'central-switch',
+        'FSPIOP-Destination': 'dfsp2'
+      },
+      'payload': {
+        'currency': 'USD',
+        'value': 0,
+        'changedDate': '2019-01-17T14:46:48.918Z'
+      }
+    },
+    'type': 'application/json',
+    'metadata': {
+      'event': {
+        'id': '5c7ff308-35be-46b0-b187-0bf4bee72cf8',
+        'type': 'notification',
+        'action': 'settlement-transfer-position-change',
+        'state': {
+          'status': 'success',
+          'code': 0,
+          'description': 'action successful'
+        }
+      },
+      'protocol.createdAt': 1547736408918
+    }
+  },
+  {
+    'id': '86ce97cf-2302-4d64-8b0e-df5ae9355dd0',
+    'from': 'central-switch',
+    'to': 'dfsp1',
+    'content': {
+      'headers': {
+        'Content-Type': 'application/json',
+        'Date': '2019-01-17T14:46:48.918Z',
+        'FSPIOP-Source': 'central-switch',
+        'FSPIOP-Destination': 'dfsp2'
+      },
+      'payload': {
+        'currency': 'USD',
+        'value': 0,
+        'changedDate': '2019-01-17T14:46:48.918Z'
+      }
+    },
+    'type': 'application/json',
+    'metadata': {
+      'event': {
+        'id': '5c7ff308-35be-46b0-b187-0bf4bee72cf8',
+        'type': 'notification',
+        'action': 'settlement-transfer-position-change',
+        'state': {
+          'status': 'success',
+          'code': 0,
+          'description': 'action successful'
+        }
+      },
+      'protocol.createdAt': 1547736408918
+    }
+  },
+  {
+    'id': '86ce97cf-2302-4d64-8b0e-df5ae9355dd0',
+    'from': 'central-switch',
+    'to': 'dfsp11',
+    'content': {
+      'headers': {
+        'Content-Type': 'application/json',
+        'Date': '2019-01-17T14:46:48.918Z',
+        'FSPIOP-Source': 'central-switch',
+        'FSPIOP-Destination': 'dfsp2'
+      },
+      'payload': {
+        'currency': 'USD',
+        'value': 0,
+        'changedDate': '2019-01-17T14:46:48.918Z'
+      }
     },
-    'payload': {
-      'currency': 'USD',
-      'value': 0,
-      'changedDate': '2019-01-17T14:46:48.918Z'
+    'type': 'application/json',
+    'metadata': {
+      'event': {
+        'id': '5c7ff308-35be-46b0-b187-0bf4bee72cf8',
+        'type': 'notification',
+        'action': 'settlement-transfer-position-change',
+        'state': {
+          'status': 'success',
+          'code': 0,
+          'description': 'action successful'
+        }
+      },
+      'protocol.createdAt': 1547736408918
     }
   },
-  'type': 'application/json',
-  'metadata': {
-    'event': {
-      'id': '5c7ff308-35be-46b0-b187-0bf4bee72cf8',
-      'type': 'notification',
-      'action': 'settlement-transfer-position-change',
-      'state': {
-        'status': 'success',
-        'code': 0,
-        'description': 'action successful'
+  {
+    'id': '86ce97cf-2302-4d64-8b0e-df5ae9355dd0',
+    'from': 'central-switch',
+    'to': 'dfsp12',
+    'content': {
+      'headers': {
+        'Content-Type': 'application/json',
+        'Date': '2019-01-17T14:46:48.918Z',
+        'FSPIOP-Source': 'central-switch',
+        'FSPIOP-Destination': 'dfsp2'
+      },
+      'payload': {
+        'currency': 'USD',
+        'value': 0,
+        'changedDate': '2019-01-17T14:46:48.918Z'
       }
     },
-    'protocol.createdAt': 1547736408918
+    'type': 'application/json',
+    'metadata': {
+      'event': {
+        'id': '5c7ff308-35be-46b0-b187-0bf4bee72cf8',
+        'type': 'notification',
+        'action': 'settlement-transfer-position-change',
+        'state': {
+          'status': 'success',
+          'code': 0,
+          'description': 'action successful'
+        }
+      },
+      'protocol.createdAt': 1547736408918
+    }
   }
-}
+
+]
+
+let count = 0
+
+const timeout = ms => new Promise(res => setTimeout(res, ms))
 
 let messageSubimtter = async (message) => {
   try {
@@ -41,10 +144,20 @@ let messageSubimtter = async (message) => {
       description: 'action successful'
     }
     await Utility.produceGeneralMessage('notification', 'settlement-transfer-position-change', message, success)
-    console.log('Message on kafka queue')
+    console.log('Message on kafka queue # ', ++count)
   } catch (err) {
     console.log('Message wasnt placed on Kaka queue : ' + err)
   }
 }
 
-messageSubimtter(message)
+let flood = async (count) => {
+  let arr = []
+  for (let i = 0; i < count; i++) {
+    for (let message of messages) {
+      // await timeout(100)
+      await messageSubimtter(message)
+    }
+  }
+}
+
+flood(20)
diff --git a/src/rxretries.js b/src/rxretries.js
@@ -0,0 +1,22 @@
+const Rx = require('rxjs')
+const { filter, flatMap, retry, delay, retryWhen, repeat, repeatWhen, catchError } = require('rxjs/operators')
+
+let arrO = Rx.from([1, 2, 3, 4, 5, 6, 7])
+
+const mapped = arrO.pipe(
+  flatMap(v => {
+    return Rx.from([2 * v])
+  }),
+  delay(1000),
+  flatMap(v => {
+    if (v === 8) return Rx.throwError('error')
+    else return Rx.from([v])
+  }),
+  catchError(e => {
+    return Rx.of(undefined)
+  })
+  )
+
+mapped.subscribe({
+  next: v => console.log(v)
+})