fix: closing a stream

Author: bangjelkoski

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcAccountPortfolioStreamV2.ts

@@ -58,14 +58,16 @@ export class IndexerGrpcAccountPortfolioStreamV2 {
       request.type = type
     }
 
-    const stream = this.client.streamAccountPortfolio(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerAccountPortfolioStreamTransformer.accountPortfolioStreamCallback(
-          response,
-        )
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamAccountPortfolio(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerAccountPortfolioStreamTransformer.accountPortfolioStreamCallback(
+            response,
+          )
+        callback(transformed)
+      },
+    )
   }
 }

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcAccountStreamV2.ts

@@ -44,12 +44,14 @@ export class IndexerGrpcAccountStreamV2 {
       InjectiveAccountsRpcPb.StreamSubaccountBalanceRequest.create()
     request.subaccountId = subaccountId
 
-    const stream = this.client.streamSubaccountBalance(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerAccountStreamTransformer.balanceStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamSubaccountBalance(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerAccountStreamTransformer.balanceStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 }

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcArchiverStreamV2.ts

@@ -45,14 +45,16 @@ export class IndexerGrpcArchiverStreamV2 {
 
     request.account = account
 
-    const stream = this.client.streamSpotAverageEntries(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerArchiverStreamTransformer.spotAverageEntriesStreamCallback(
-          response,
-        )
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamSpotAverageEntries(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerArchiverStreamTransformer.spotAverageEntriesStreamCallback(
+            response,
+          )
+        callback(transformed)
+      },
+    )
   }
 }

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcAuctionStreamV2.ts

@@ -36,12 +36,13 @@ export class IndexerGrpcAuctionStreamV2 {
 
     const request = InjectiveAuctionRpcPb.StreamBidsRequest.create()
 
-    const stream = this.client.streamBids(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerAuctionStreamTransformer.bidsStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) => this.client.streamBids(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerAuctionStreamTransformer.bidsStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 }

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcDerivativesStreamV2.spec.ts

@@ -0,0 +1,40 @@
+import { vi } from 'vitest'
+import { Network, getNetworkEndpoints } from '@injectivelabs/networks'
+import { IndexerGrpcDerivativesStreamV2 } from './IndexerGrpcDerivativesStreamV2.js'
+
+const endpoints = getNetworkEndpoints(Network.MainnetSentry)
+const mockMarketId =
+  '0x17ef48032cb24375ba7c2e39f384e56433bcab20cbee9a7357e4cba2eb00abe6'
+
+describe('IndexerGrpcDerivativesStreamV2', () => {
+  let derivativesStream: IndexerGrpcDerivativesStreamV2
+
+  beforeEach(() => {
+    derivativesStream = new IndexerGrpcDerivativesStreamV2(endpoints.indexer)
+  })
+
+  describe('streamOrderbooksV2', () => {
+    it('should pass an abort signal to the generated stream and abort it on unsubscribe', () => {
+      const streamOrderbookV2 = vi.fn().mockReturnValue({
+        responses: (async function* emptyStream() {})(),
+      })
+
+      ;(derivativesStream as any).client = { streamOrderbookV2 }
+
+      const subscription = derivativesStream.streamOrderbooksV2({
+        marketIds: [mockMarketId],
+        callback: () => {},
+      })
+
+      const [request, options] = streamOrderbookV2.mock.calls[0]
+
+      expect(request.marketIds).toEqual([mockMarketId])
+      expect(options.abort).toBeDefined()
+      expect(options.abort.aborted).toBe(false)
+
+      subscription.unsubscribe()
+
+      expect(options.abort.aborted).toBe(true)
+    })
+  })
+})

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcDerivativesStreamV2.ts

@@ -106,13 +106,15 @@ export class IndexerGrpcDerivativesStreamV2 {
       request.orderSide = orderSide
     }
 
-    const stream = this.client.streamOrders(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.ordersStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamOrders(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.ordersStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -174,13 +176,17 @@ export class IndexerGrpcDerivativesStreamV2 {
       request.executionTypes = executionTypes
     }
 
-    const stream = this.client.streamOrdersHistory(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.orderHistoryStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamOrdersHistory(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.orderHistoryStreamCallback(
+            response,
+          )
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -255,13 +261,15 @@ export class IndexerGrpcDerivativesStreamV2 {
       }
     }
 
-    const stream = this.client.streamTrades(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.tradesStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamTrades(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.tradesStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -302,13 +310,15 @@ export class IndexerGrpcDerivativesStreamV2 {
       request.subaccountId = subaccountId
     }
 
-    const stream = this.client.streamPositions(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.positionStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamPositions(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.positionStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -335,11 +345,13 @@ export class IndexerGrpcDerivativesStreamV2 {
       request.marketIds = marketIds
     }
 
-    const stream = this.client.streamMarket(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      callback(response)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamMarket(request, { abort: abortSignal }),
+      (response) => {
+        callback(response)
+      },
+    )
   }
 
   /**
@@ -366,13 +378,15 @@ export class IndexerGrpcDerivativesStreamV2 {
       InjectiveDerivativeExchangeRpcPb.StreamOrderbookV2Request.create()
     request.marketIds = marketIds
 
-    const stream = this.client.streamOrderbookV2(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.orderbookV2StreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamOrderbookV2(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.orderbookV2StreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -400,15 +414,17 @@ export class IndexerGrpcDerivativesStreamV2 {
 
     request.marketIds = marketIds
 
-    const stream = this.client.streamOrderbookUpdate(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.orderbookUpdateStreamCallback(
-          response,
-        )
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamOrderbookUpdate(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.orderbookUpdateStreamCallback(
+            response,
+          )
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -449,12 +465,14 @@ export class IndexerGrpcDerivativesStreamV2 {
       request.subaccountId = subaccountId
     }
 
-    const stream = this.client.streamPositionsV2(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        IndexerDerivativeStreamTransformer.positionV2StreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamPositionsV2(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          IndexerDerivativeStreamTransformer.positionV2StreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 }

packages/sdk-ts/src/client/indexer/grpc_stream/streamV2/IndexerGrpcExplorerStreamV2.ts

@@ -45,13 +45,15 @@ export class IndexerGrpcExplorerStreamV2 {
     }
 
     const request = InjectiveExplorerRpcPb.StreamBlocksRequest.create()
-    const stream = this.client.streamBlocks(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        ExplorerStreamTransformer.blocksStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamBlocks(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          ExplorerStreamTransformer.blocksStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -69,13 +71,15 @@ export class IndexerGrpcExplorerStreamV2 {
     }
 
     const request = InjectiveExplorerRpcPb.StreamBlocksRequest.create()
-    const stream = this.client.streamBlocks(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        ExplorerStreamTransformer.blocksWithTxsStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) =>
+        this.client.streamBlocks(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          ExplorerStreamTransformer.blocksWithTxsStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 
   /**
@@ -93,12 +97,13 @@ export class IndexerGrpcExplorerStreamV2 {
     }
 
     const request = InjectiveExplorerRpcPb.StreamTxsRequest.create()
-    const stream = this.client.streamTxs(request)
-
-    return createStreamSubscriptionV2(stream, (response) => {
-      const transformed =
-        ExplorerStreamTransformer.transactionsStreamCallback(response)
-      callback(transformed)
-    })
+    return createStreamSubscriptionV2(
+      (abortSignal) => this.client.streamTxs(request, { abort: abortSignal }),
+      (response) => {
+        const transformed =
+          ExplorerStreamTransformer.transactionsStreamCallback(response)
+        callback(transformed)
+      },
+    )
   }
 }