refactor(client): Inline PushPipeline#pull method (#1553)

Inlined `PushPipeline#pull` method which was just a delegated to the `pull` function in `PushBuffer`. The utility function works both with `PushPipeline` and `PushBuffer` it would make sense to have the delegator method in both classes, or in neither class. Maybe the best approach is to keep the pipeline and buffer classes as simple as possible.

Removed also `PushBuffer#pull` and `PushBuffer#filter` methods which were not used.

Author: teogeb

packages/client/src/subscribe/Resends.ts

@@ -10,6 +10,7 @@ import { StorageNodeRegistry } from '../registry/StorageNodeRegistry'
 import { StreamStorageRegistry } from '../registry/StreamStorageRegistry'
 import { forEach, map } from '../utils/GeneratorUtils'
 import { LoggerFactory } from '../utils/LoggerFactory'
+import { pull } from '../utils/PushBuffer'
 import { PushPipeline } from '../utils/PushPipeline'
 import { createQueryString, fetchHttpStream } from '../utils/utils'
 import { MessagePipelineFactory } from './MessagePipelineFactory'
@@ -190,8 +191,9 @@ export class Resends {
         setImmediate(async () => {
             let count = 0
             const messages = map(lines, (line: string) => StreamMessage.deserialize(line))
-            await messageStream.pull(
-                forEach(messages, () => count++)
+            await pull(
+                forEach(messages, () => count++),
+                messageStream
             )
             this.logger.debug('Finished resend', { loggerIdx: traceId, messageCount: count })
         })

packages/client/src/utils/PushBuffer.ts

@@ -1,5 +1,4 @@
 import { Gate } from './Gate'
-import * as G from './GeneratorUtils'
 import { StreamrClientError } from '../StreamrClientError'
 
 export const DEFAULT_BUFFER_SIZE = 256
@@ -79,12 +78,6 @@ export class PushBuffer<T> implements IPushBuffer<T> {
         return this.writeGate.check()
     }
 
-    filter(fn: G.GeneratorFilter<T>): PushBuffer<unknown> {
-        const p = new PushBuffer(this.bufferSize)
-        pull(G.filter(this, fn), p)
-        return p
-    }
-
     private updateWriteGate(): void {
         this.writeGate.setOpenState(!this.isFull())
     }
@@ -215,18 +208,6 @@ export class PushBuffer<T> implements IPushBuffer<T> {
         return this.iterator.next()
     }
 
-    async pull(src: AsyncGenerator<T>): Promise<void> {
-        try {
-            for await (const v of src) {
-                const ok = await this.push(v)
-                if (!ok || !this.isWritable()) { break }
-            }
-        } catch (err) {
-            // this.endWrite(err)
-        }
-        this.endWrite()
-    }
-
     [Symbol.asyncIterator](): this {
         if (this.isIterating) {
             // @ts-expect-error ts can't do this.constructor properly

packages/client/src/utils/PushPipeline.ts

@@ -1,4 +1,4 @@
-import { IPushBuffer, PushBuffer, DEFAULT_BUFFER_SIZE, pull } from './PushBuffer'
+import { IPushBuffer, PushBuffer, DEFAULT_BUFFER_SIZE } from './PushBuffer'
 import * as G from './GeneratorUtils'
 import { Pipeline, PipelineTransform } from './Pipeline'
 
@@ -26,10 +26,6 @@ export class PushPipeline<InType, OutType = InType> extends Pipeline<InType, Out
         return super.filter(fn) as PushPipeline<InType, OutType>
     }
 
-    pull(source: AsyncGenerator<InType>): Promise<void> {
-        return pull(source, this)
-    }
-
     // wrapped PushBuffer methods below here
 
     async push(item: InType | Error): Promise<boolean> {

packages/client/test/test-utils/utils.ts

@@ -5,7 +5,10 @@ import { MAX_PARTITION_COUNT, StreamMessage, StreamPartID, StreamPartIDUtils } f
 import { fastPrivateKey, fetchPrivateKeyWithGas } from '@streamr/test-utils'
 import { EthereumAddress, Logger, merge, wait, waitForCondition } from '@streamr/utils'
 import crypto from 'crypto'
+import { once } from 'events'
+import express, { Request, Response } from 'express'
 import { mock } from 'jest-mock-extended'
+import { AddressInfo } from 'net'
 import { DependencyContainer } from 'tsyringe'
 import { Authentication, createPrivateKeyAuthentication } from '../../src/Authentication'
 import { StreamrClientConfig } from '../../src/Config'
@@ -223,3 +226,23 @@ export const waitForCalls = async (mockFunction: jest.Mock<any>, n: number): Pro
         return `Timeout while waiting for calls: got ${mockFunction.mock.calls.length} out of ${n}`
     })
 }
+
+export const startTestServer = async (
+    endpoint: string,
+    onRequest: (req: Request, res: Response) => Promise<void>
+): Promise<{ url: string, stop: () => Promise<void> }> => {
+    const app = express()
+    app.get(endpoint, async (req, res) => {
+        await onRequest(req, res)
+    })
+    const server = app.listen()
+    await once(server, 'listening')
+    const port = (server.address() as AddressInfo).port
+    return {
+        url: `http://localhost:${port}`,
+        stop: async () => {
+            server.close()
+            await once(server, 'close')
+        }
+    }
+}

packages/client/test/unit/Pipeline.test.ts

@@ -643,8 +643,7 @@ describe('Pipeline', () => {
                     yield* s
                 })
                 pipeline.onFinally.listen(onFinally)
-
-                pipeline.pull(generate())
+                pull(generate(), pipeline)
                 return pipeline
             })
 
@@ -654,11 +653,10 @@ describe('Pipeline', () => {
                     yield* s
                 })
                 pipeline.onFinally.listen(onFinally)
-
-                pipeline.pull((async function* generateError() {
+                pull((async function* generateError() {
                     yield* generate()
                     throw err
-                }()))
+                }()), pipeline)
                 const received: number[] = []
                 await expect(async () => {
                     for await (const msg of pipeline) {
@@ -675,11 +673,10 @@ describe('Pipeline', () => {
                     yield* s
                 })
                 pipeline.onFinally.listen(onFinally)
-
                 // eslint-disable-next-line require-yield
-                pipeline.pull((async function* generateError() {
+                pull((async function* generateError() {
                     throw err
-                }()))
+                }()), pipeline)
                 const received: any[] = []
                 await expect(async () => {
                     for await (const msg of pipeline) {

packages/client/test/unit/PushPipeline.test.ts

@@ -2,6 +2,7 @@ import { MessageID, StreamMessage, toStreamID } from '@streamr/protocol'
 import { collect, toEthereumAddress, wait } from '@streamr/utils'
 import { Authentication } from '../../src/Authentication'
 import { createSignedMessage } from '../../src/publish/MessageFactory'
+import { pull } from '../../src/utils/PushBuffer'
 import { PushPipeline } from '../../src/utils/PushPipeline'
 import { counterId, instanceId } from '../../src/utils/utils'
 import { LeaksDetector } from '../test-utils/LeaksDetector'
@@ -61,11 +62,10 @@ describe('PushPipeline', () => {
         const s = new PushPipeline<StreamMessage>()
         leaksDetector.add(instanceId(s), s)
         const received: StreamMessage[] = []
-        s.pull((async function* g() {
+        pull((async function* g() {
             yield streamMessage
-
             throw err
-        }()))
+        }()), s)
 
         await expect(async () => {
             for await (const msg of s) {
@@ -95,9 +95,9 @@ describe('PushPipeline', () => {
             throw error
         })
         // eslint-disable-next-line require-yield
-        s.pull((async function* g() {
+        pull((async function* g() {
             throw err
-        }()))
+        }()), s)
 
         await expect(async () => {
             for await (const msg of s) {