deps: update it-stream-types (#456)

Updates stream types and abortable iterator

Author: achingbrain

.gitignore

@@ -6,3 +6,4 @@ dist
 node_modules
 package-lock.json
 yarn.lock
+.vscode

package.json

@@ -144,7 +144,7 @@
   "dependencies": {
     "@libp2p/crypto": "^1.0.4",
     "@libp2p/interface-address-manager": "^2.0.0",
-    "@libp2p/interface-connection": "^4.0.0",
+    "@libp2p/interface-connection": "^5.0.1",
     "@libp2p/interface-connection-manager": "^2.0.0",
     "@libp2p/interface-dht": "^2.0.0",
     "@libp2p/interface-metrics": "^4.0.0",
@@ -160,7 +160,7 @@
     "@libp2p/record": "^3.0.0",
     "@libp2p/topology": "^4.0.0",
     "@multiformats/multiaddr": "^12.0.0",
-    "abortable-iterator": "^4.0.2",
+    "abortable-iterator": "^5.0.1",
     "any-signal": "^4.1.1",
     "datastore-core": "^9.0.1",
     "hashlru": "^2.3.0",
@@ -174,7 +174,7 @@
     "it-merge": "^3.0.0",
     "it-parallel": "^3.0.0",
     "it-pipe": "^3.0.0",
-    "it-stream-types": "^1.0.4",
+    "it-stream-types": "^2.0.1",
     "it-take": "^3.0.1",
     "k-bucket": "^5.1.0",
     "multiformats": "^11.0.0",
@@ -187,7 +187,7 @@
     "varint": "^6.0.0"
   },
   "devDependencies": {
-    "@libp2p/interface-mocks": "^9.0.0",
+    "@libp2p/interface-mocks": "^10.0.2",
     "@libp2p/peer-id-factory": "^2.0.0",
     "@libp2p/peer-store": "^7.0.0",
     "@types/lodash.random": "^3.2.6",

src/network.ts

@@ -16,7 +16,7 @@ import type { PeerId } from '@libp2p/interface-peer-id'
 import type { AbortOptions } from '@libp2p/interfaces'
 import type { Startable } from '@libp2p/interfaces/startable'
 import type { Logger } from '@libp2p/logger'
-import type { Duplex } from 'it-stream-types'
+import type { Duplex, Source } from 'it-stream-types'
 import type { PeerInfo } from '@libp2p/interface-peer-info'
 import type { Stream } from '@libp2p/interface-connection'
 import { abortableDuplex } from 'abortable-iterator'
@@ -149,7 +149,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable {
   /**
    * Write a message to the given stream
    */
-  async _writeMessage (stream: Duplex<Uint8ArrayList, Uint8ArrayList | Uint8Array>, msg: Uint8Array | Uint8ArrayList, options: AbortOptions): Promise<void> {
+  async _writeMessage (stream: Duplex<AsyncGenerator<Uint8ArrayList>, Source<Uint8ArrayList | Uint8Array>>, msg: Uint8Array | Uint8ArrayList, options: AbortOptions): Promise<void> {
     if (options.signal != null) {
       stream = abortableDuplex(stream, options.signal)
     }
@@ -167,7 +167,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable {
    * If no response is received after the specified timeout
    * this will error out.
    */
-  async _writeReadMessage (stream: Duplex<Uint8ArrayList, Uint8ArrayList | Uint8Array>, msg: Uint8Array | Uint8ArrayList, options: AbortOptions): Promise<Message> {
+  async _writeReadMessage (stream: Duplex<AsyncGenerator<Uint8ArrayList>, Source<Uint8ArrayList | Uint8Array>>, msg: Uint8Array | Uint8ArrayList, options: AbortOptions): Promise<Message> {
     if (options.signal != null) {
       stream = abortableDuplex(stream, options.signal)
     }

test/network.spec.ts

@@ -12,7 +12,7 @@ import { mockStream } from '@libp2p/interface-mocks'
 import type { DualKadDHT } from '../src/dual-kad-dht.js'
 import type { Connection } from '@libp2p/interface-connection'
 import type { PeerId } from '@libp2p/interface-peer-id'
-import type { Sink } from 'it-stream-types'
+import type { Sink, Source } from 'it-stream-types'
 import { Uint8ArrayList } from 'uint8arraylist'
 import map from 'it-map'
 import type { Multiaddr } from '@multiformats/multiaddr'
@@ -68,13 +68,13 @@ describe('Network', () => {
               async (source) => all(source)
             )
 
-            const source = (function * () {
+            const source = (async function * () {
               const array = data
 
               yield * array
             })()
 
-            const sink: Sink<Uint8ArrayList | Uint8Array> = async source => {
+            const sink: Sink<Source<Uint8ArrayList | Uint8Array>, Promise<void>> = async source => {
               const res = await pipe(
                 source,
                 (source) => lp.decode(source),

test/rpc/index.node.ts

@@ -18,7 +18,7 @@ import { PeerRouting } from '../../src/peer-routing/index.js'
 import type { Validators } from '@libp2p/interface-dht'
 import type { Datastore } from 'interface-datastore'
 import { RoutingTable } from '../../src/routing-table/index.js'
-import type { Duplex } from 'it-stream-types'
+import type { Duplex, Source } from 'it-stream-types'
 import { mockStream } from '@libp2p/interface-mocks'
 import { start } from '@libp2p/interfaces/startable'
 import { Uint8ArrayList } from 'uint8arraylist'
@@ -85,8 +85,10 @@ describe('rpc', () => {
       (source) => all(source)
     )
 
-    const duplexStream: Duplex<Uint8ArrayList, Uint8ArrayList | Uint8Array> = {
-      source,
+    const duplexStream: Duplex<AsyncGenerator<Uint8ArrayList>, Source<Uint8ArrayList | Uint8Array>, Promise<void>> = {
+      source: (async function * () {
+        yield * source
+      })(),
       sink: async (source) => {
         const res = await pipe(
           source,