fix!: update stream types (#316)

* deps!: update stream types

libp2p streams are now explicit about the types of sync/sources they provide, showing that they are
`AsyncGenerator`s and not just `AsyncIterable`s.

Refs: achingbrain/it-stream-types#45

BREAKING CHANGE: the type of the source/sink properties have changed

* fix: interop

Signed-off-by: Marin Petrunic <marin.petrunic@gmail.com>

---------

Signed-off-by: Marin Petrunic <marin.petrunic@gmail.com>
Co-authored-by: Marin Petrunic <marin.petrunic@gmail.com>

Author: achingbrain

.gitignore

@@ -17,3 +17,4 @@ lerna-debug.log*
 coverage/*
 package-lock.json
 yarn.lock
+.vscode

package.json

@@ -69,7 +69,7 @@
   },
   "dependencies": {
     "@libp2p/crypto": "^1.0.11",
-    "@libp2p/interface-connection-encrypter": "^3.0.5",
+    "@libp2p/interface-connection-encrypter": "^4.0.0",
     "@libp2p/interface-keys": "^1.0.6",
     "@libp2p/interface-metrics": "^4.0.4",
     "@libp2p/interface-peer-id": "^2.0.0",
@@ -79,31 +79,31 @@
     "@stablelib/hkdf": "^1.0.1",
     "@stablelib/sha256": "^1.0.1",
     "@stablelib/x25519": "^1.0.3",
-    "it-length-prefixed": "^8.0.2",
+    "it-length-prefixed": "^9.0.1",
     "it-pair": "^2.0.2",
-    "it-pb-stream": "^3.2.0",
-    "it-pipe": "^2.0.3",
-    "it-stream-types": "^1.0.4",
+    "it-pb-stream": "^4.0.1",
+    "it-pipe": "^3.0.1",
+    "it-stream-types": "^2.0.1",
     "protons-runtime": "^5.0.0",
     "uint8arraylist": "^2.3.2",
     "uint8arrays": "^4.0.2"
   },
   "devDependencies": {
-    "@libp2p/daemon-client": "^5.0.0",
-    "@libp2p/daemon-server": "^4.0.1",
-    "@libp2p/interface-connection-encrypter-compliance-tests": "^4.0.0",
-    "@libp2p/interop": "^7.0.3",
-    "@libp2p/mplex": "^7.0.0",
+    "@chainsafe/libp2p-yamux": "^4.0.1",
+    "@libp2p/daemon-client": "^6.0.3",
+    "@libp2p/daemon-server": "^5.0.2",
+    "@libp2p/interface-connection-encrypter-compliance-tests": "^5.0.0",
+    "@libp2p/interop": "^8.0.1",
     "@libp2p/peer-id-factory": "^2.0.0",
-    "@libp2p/tcp": "^6.0.2",
+    "@libp2p/tcp": "^7.0.0",
     "@multiformats/multiaddr": "^12.1.0",
     "aegir": "^38.1.7",
     "benchmark": "^2.1.4",
     "execa": "^7.0.0",
-    "go-libp2p": "^0.0.6",
+    "go-libp2p": "^1.0.3",
     "iso-random-stream": "^2.0.2",
-    "libp2p": "^0.43.2",
-    "mkdirp": "^2.0.0",
+    "libp2p": "0.44.0-59f34a21",
+    "mkdirp": "^3.0.0",
     "p-defer": "^4.0.0",
     "protons": "^7.0.0",
     "sinon": "^15.0.0",

src/crypto/streaming.ts

@@ -7,7 +7,7 @@ import { NOISE_MSG_MAX_LENGTH_BYTES, NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG } fr
 import { uint16BEEncode } from '../encoder.js'
 
 // Returns generator that encrypts payload from the user
-export function encryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<Uint8Array> {
+export function encryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<AsyncIterable<Uint8Array>> {
   return async function * (source) {
     for await (const chunk of source) {
       for (let i = 0; i < chunk.length; i += NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG) {
@@ -27,7 +27,7 @@ export function encryptStream (handshake: IHandshake, metrics?: MetricsRegistry)
 }
 
 // Decrypt received payload to the user
-export function decryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<Uint8ArrayList, Uint8Array> {
+export function decryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<AsyncIterable<Uint8ArrayList>, AsyncIterable<Uint8Array>> {
   return async function * (source) {
     for await (const chunk of source) {
       for (let i = 0; i < chunk.length; i += NOISE_MSG_MAX_LENGTH_BYTES) {

src/noise.ts

@@ -4,7 +4,7 @@ import { pbStream, ProtobufStream } from 'it-pb-stream'
 import { duplexPair } from 'it-pair/duplex'
 import { pipe } from 'it-pipe'
 import { decode } from 'it-length-prefixed'
-import type { Duplex } from 'it-stream-types'
+import type { Duplex, Source } from 'it-stream-types'
 import type { bytes } from './@types/basic.js'
 import type { IHandshake } from './@types/handshake-interface.js'
 import type { INoiseConnection, KeyPair } from './@types/libp2p.js'
@@ -66,11 +66,11 @@ export class Noise implements INoiseConnection {
    * Encrypt outgoing data to the remote party (handshake as initiator)
    *
    * @param {PeerId} localPeer - PeerId of the receiving peer
-   * @param {Duplex<Uint8Array>} connection - streaming iterable duplex that will be encrypted
+   * @param {Duplex<AsyncGenerator<Uint8Array>, AsyncIterable<Uint8Array>, Promise<void>>} connection - streaming iterable duplex that will be encrypted
    * @param {PeerId} remotePeer - PeerId of the remote peer. Used to validate the integrity of the remote peer.
    * @returns {Promise<SecuredConnection>}
    */
-  public async secureOutbound (localPeer: PeerId, connection: Duplex<Uint8Array>, remotePeer?: PeerId): Promise<SecuredConnection<NoiseExtensions>> {
+  public async secureOutbound (localPeer: PeerId, connection: Duplex<AsyncGenerator<Uint8Array>, AsyncIterable<Uint8Array>, Promise<void>>, remotePeer?: PeerId): Promise<SecuredConnection<NoiseExtensions>> {
     const wrappedConnection = pbStream(
       connection,
       {
@@ -98,11 +98,11 @@ export class Noise implements INoiseConnection {
    * Decrypt incoming data (handshake as responder).
    *
    * @param {PeerId} localPeer - PeerId of the receiving peer.
-   * @param {Duplex<Uint8Array>} connection - streaming iterable duplex that will be encryption.
+   * @param {Duplex<AsyncGenerator<Uint8Array>, AsyncIterable<Uint8Array>, Promise<void>>} connection - streaming iterable duplex that will be encryption.
    * @param {PeerId} remotePeer - optional PeerId of the initiating peer, if known. This may only exist during transport upgrades.
    * @returns {Promise<SecuredConnection>}
    */
-  public async secureInbound (localPeer: PeerId, connection: Duplex<Uint8Array>, remotePeer?: PeerId): Promise<SecuredConnection<NoiseExtensions>> {
+  public async secureInbound (localPeer: PeerId, connection: Duplex<AsyncGenerator<Uint8Array>, AsyncIterable<Uint8Array>, Promise<void>>, remotePeer?: PeerId): Promise<SecuredConnection<NoiseExtensions>> {
     const wrappedConnection = pbStream(
       connection,
       {
@@ -171,9 +171,9 @@ export class Noise implements INoiseConnection {
   }
 
   private async createSecureConnection (
-    connection: ProtobufStream,
+    connection: ProtobufStream<Duplex<AsyncGenerator<Uint8Array>, AsyncIterable<Uint8Array>, Promise<void>>>,
     handshake: IHandshake
-  ): Promise<Duplex<Uint8Array>> {
+  ): Promise<Duplex<AsyncGenerator<Uint8Array>, Source<Uint8Array>, Promise<void>>> {
     // Create encryption box/unbox wrapper
     const [secure, user] = duplexPair<Uint8Array>()
     const network = connection.unwrap()
@@ -182,7 +182,7 @@ export class Noise implements INoiseConnection {
       secure, // write to wrapper
       encryptStream(handshake, this.metrics), // encrypt data + prefix with message length
       network, // send to the remote peer
-      decode({ lengthDecoder: uint16BEDecode }), // read message length prefix
+      (source) => decode(source, { lengthDecoder: uint16BEDecode }), // read message length prefix
       decryptStream(handshake, this.metrics), // decrypt the incoming data
       secure // pipe to the wrapper
     )

test/interop.ts

@@ -9,7 +9,7 @@ import { path as p2pd } from 'go-libp2p'
 import { execa } from 'execa'
 import pDefer from 'p-defer'
 import { logger } from '@libp2p/logger'
-import { mplex } from '@libp2p/mplex'
+import { yamux } from '@chainsafe/libp2p-yamux'
 import fs from 'fs'
 import { unmarshalPrivateKey } from '@libp2p/crypto/keys'
 import type { PeerId } from '@libp2p/interface-peer-id'
@@ -77,7 +77,7 @@ async function createJsPeer (options: SpawnOptions): Promise<Daemon> {
       listen: ['/ip4/0.0.0.0/tcp/0']
     },
     transports: [tcp()],
-    streamMuxers: [mplex()],
+    streamMuxers: [yamux()],
     connectionEncryption: [noise()]
   }