SyncAgent - refactor out the promise queue (#256)

packages/sdk/src/sync-agent/river-connection/riverConnection.ts

@@ -4,17 +4,14 @@ import { Observable } from '../../observable/observable'
 import { RiverNodeUrls } from './models/riverNodeUrls'
 import { Store } from '../../store/store'
 import { dlogger } from '@river-build/dlog'
+import { PromiseQueue } from '../utils/promiseQueue'
 
 const logger = dlogger('csb:riverConnection')
 
 export class RiverConnection {
     rpcClient: Observable<StreamRpcClient | undefined>
     nodeUrls: RiverNodeUrls
-    queue: {
-        resolve: (value: any) => void
-        reject: (reason?: any) => void
-        fn: (rpcClient: StreamRpcClient) => Promise<any>
-    }[] = []
+    rpcClientQueue = new PromiseQueue<StreamRpcClient>()
 
     constructor(store: Store, riverRegistryDapp: RiverRegistry, retryParams?: RetryParams) {
         this.rpcClient = new Observable<StreamRpcClient | undefined>(undefined)
@@ -23,13 +20,11 @@ export class RiverConnection {
             (value) => {
                 if (value.data.urls) {
                     logger.log('RiverConnection: setting rpcClient', value.data.urls)
-                    this.rpcClient.set(
-                        makeStreamRpcClient(value.data.urls, retryParams, () =>
-                            riverRegistryDapp.getOperationalNodeUrls(),
-                        ),
+                    const client = makeStreamRpcClient(value.data.urls, retryParams, () =>
+                        riverRegistryDapp.getOperationalNodeUrls(),
                     )
-                    // New rpcClient is available, resolve all queued requests
-                    this.flushQueue()
+                    this.rpcClient.set(client)
+                    this.rpcClientQueue.flush(client) // New rpcClient is available, resolve all queued requests
                 } else {
                     this.rpcClient.set(undefined)
                 }
@@ -44,19 +39,7 @@ export class RiverConnection {
             return fn(rpcClient)
         } else {
             // Enqueue the request if rpcClient is not available
-            return new Promise<T>((resolve, reject) => {
-                this.queue.push({ resolve, reject, fn })
-            })
-        }
-    }
-
-    private flushQueue() {
-        if (this.rpcClient.value && this.queue.length) {
-            logger.log('RiverConnection: flushing rpc queue', this.queue.length)
-            while (this.queue.length > 0) {
-                const { resolve, reject, fn } = this.queue.shift()!
-                fn(this.rpcClient.value).then(resolve).catch(reject)
-            }
+            return this.rpcClientQueue.enqueue(fn)
         }
     }
 }

packages/sdk/src/sync-agent/utils/promiseQueue.ts

@@ -0,0 +1,27 @@
+import { dlogger } from '@river-build/dlog'
+
+const logger = dlogger('csb:promisequeue')
+
+export class PromiseQueue<T> {
+    private queue: {
+        resolve: (value: any) => void
+        reject: (reason?: any) => void
+        fn: (object: T) => Promise<any>
+    }[] = []
+
+    enqueue<Q>(fn: (object: T) => Promise<Q>) {
+        return new Promise<Q>((resolve, reject) => {
+            this.queue.push({ resolve, reject, fn })
+        })
+    }
+
+    flush(object: T) {
+        if (this.queue.length) {
+            logger.log('RiverConnection: flushing rpc queue', this.queue.length)
+            while (this.queue.length > 0) {
+                const { resolve, reject, fn } = this.queue.shift()!
+                fn(object).then(resolve).catch(reject)
+            }
+        }
+    }
+}