feat!: streams as EventTargets

packages/connection-encrypter-plaintext/package.json

@@ -48,7 +48,7 @@
     "@libp2p/crypto": "^5.1.7",
     "@libp2p/interface": "^2.10.5",
     "@libp2p/peer-id": "^5.1.8",
-    "it-protobuf-stream": "^2.0.2",
+    "@libp2p/utils": "^6.7.1",
     "protons-runtime": "^5.5.0",
     "uint8arraylist": "^2.4.8",
     "uint8arrays": "^5.1.0"
@@ -57,7 +57,6 @@
     "@libp2p/crypto": "^5.1.7",
     "@libp2p/logger": "^5.1.21",
     "aegir": "^47.0.14",
-    "it-pair": "^2.0.6",
     "protons": "^7.6.1",
     "sinon": "^20.0.0"
   },

packages/connection-encrypter-plaintext/src/index.ts

@@ -24,10 +24,10 @@
 import { publicKeyFromRaw } from '@libp2p/crypto/keys'
 import { UnexpectedPeerError, InvalidCryptoExchangeError, serviceCapabilities, ProtocolError } from '@libp2p/interface'
 import { peerIdFromPublicKey } from '@libp2p/peer-id'
-import { pbStream } from 'it-protobuf-stream'
+import { pbStream } from '@libp2p/utils'
 import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
 import { Exchange, KeyType } from './pb/proto.js'
-import type { ComponentLogger, Logger, MultiaddrConnection, ConnectionEncrypter, SecuredConnection, PrivateKey, SecureConnectionOptions, SecurableStream } from '@libp2p/interface'
+import type { ComponentLogger, Logger, MultiaddrConnection, ConnectionEncrypter, SecuredConnection, PrivateKey, SecureConnectionOptions, MessageStream } from '@libp2p/interface'
 
 const PROTOCOL = '/plaintext/2.0.0'
 
@@ -52,18 +52,18 @@ class Plaintext implements ConnectionEncrypter {
     '@libp2p/connection-encryption'
   ]
 
-  async secureInbound<Stream extends SecurableStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureInbound<Stream extends MessageStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     return this._encrypt(conn, options)
   }
 
-  async secureOutbound<Stream extends SecurableStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureOutbound<Stream extends MessageStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     return this._encrypt(conn, options)
   }
 
   /**
    * Encrypt connection
    */
-  async _encrypt<Stream extends SecurableStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async _encrypt<Stream extends MessageStream = MultiaddrConnection>(conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     const log = conn.log?.newScope('plaintext') ?? this.log
     const pb = pbStream(conn).pb(Exchange)
 

packages/connection-encrypter-plaintext/test/index.spec.ts

@@ -3,8 +3,8 @@
 import { generateKeyPair } from '@libp2p/crypto/keys'
 import { defaultLogger } from '@libp2p/logger'
 import { peerIdFromPrivateKey } from '@libp2p/peer-id'
+import { streamPair } from '@libp2p/utils'
 import { expect } from 'aegir/chai'
-import { duplexPair } from 'it-pair/duplex'
 import sinon from 'sinon'
 import { plaintext } from '../src/index.js'
 import type { ConnectionEncrypter, PeerId } from '@libp2p/interface'
@@ -36,7 +36,7 @@ describe('plaintext', () => {
   })
 
   it('should verify the public key and id match', async () => {
-    const [inbound, outbound] = duplexPair<any>()
+    const [inbound, outbound] = await streamPair()
 
     await Promise.all([
       encrypter.secureInbound(inbound),
@@ -57,7 +57,7 @@ describe('plaintext', () => {
       logger: defaultLogger()
     })
 
-    const [inbound, outbound] = duplexPair<any>()
+    const [inbound, outbound] = await streamPair()
 
     await expect(Promise.all([
       encrypter.secureInbound(inbound),

packages/connection-encrypter-tls/package.json

@@ -44,22 +44,21 @@
     "@libp2p/crypto": "^5.1.7",
     "@libp2p/interface": "^2.10.5",
     "@libp2p/peer-id": "^5.1.8",
+    "@libp2p/utils": "^6.7.1",
     "@peculiar/asn1-schema": "^2.3.15",
     "@peculiar/asn1-x509": "^2.3.15",
     "@peculiar/webcrypto": "^1.5.0",
     "@peculiar/x509": "^1.12.3",
     "asn1js": "^3.0.6",
-    "it-queueless-pushable": "^2.0.1",
-    "it-stream-types": "^2.0.2",
     "protons-runtime": "^5.5.0",
     "uint8arraylist": "^2.4.8",
     "uint8arrays": "^5.1.0"
   },
   "devDependencies": {
     "@libp2p/logger": "^5.1.21",
     "aegir": "^47.0.14",
-    "it-pair": "^2.0.6",
     "protons": "^7.6.1",
+    "race-event": "^1.6.1",
     "sinon": "^20.0.0",
     "sinon-ts": "^2.0.0"
   },

packages/connection-encrypter-tls/src/tls.browser.ts

@@ -1,8 +1,6 @@
 import { serviceCapabilities } from '@libp2p/interface'
 import { PROTOCOL } from './index.js'
-import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, SecureConnectionOptions } from '@libp2p/interface'
-import type { Duplex } from 'it-stream-types'
-import type { Uint8ArrayList } from 'uint8arraylist'
+import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, SecureConnectionOptions, MessageStream } from '@libp2p/interface'
 
 export class TLS implements ConnectionEncrypter {
   public protocol: string = PROTOCOL
@@ -17,11 +15,11 @@ export class TLS implements ConnectionEncrypter {
     '@libp2p/connection-encryption'
   ]
 
-  async secureInbound <Stream extends Duplex<AsyncGenerator<Uint8Array | Uint8ArrayList>> = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureInbound <Stream extends MessageStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
     throw new Error('TLS encryption is not possible in browsers')
   }
 
-  async secureOutbound <Stream extends Duplex<AsyncGenerator<Uint8Array | Uint8ArrayList>> = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureOutbound <Stream extends MessageStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
     throw new Error('TLS encryption is not possible in browsers')
   }
 }

packages/connection-encrypter-tls/src/tls.ts

@@ -21,10 +21,10 @@
 import { TLSSocket, connect } from 'node:tls'
 import { InvalidCryptoExchangeError, serviceCapabilities } from '@libp2p/interface'
 import { HandshakeTimeoutError } from './errors.js'
-import { generateCertificate, verifyPeerCertificate, itToStream, streamToIt } from './utils.js'
+import { generateCertificate, verifyPeerCertificate, toNodeDuplex, toMessageStream } from './utils.js'
 import { PROTOCOL } from './index.js'
 import type { TLSComponents } from './index.js'
-import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, Logger, SecureConnectionOptions, CounterGroup, StreamMuxerFactory, SecurableStream } from '@libp2p/interface'
+import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, Logger, SecureConnectionOptions, CounterGroup, StreamMuxerFactory, MessageStream } from '@libp2p/interface'
 import type { TLSSocketOptions } from 'node:tls'
 
 export class TLS implements ConnectionEncrypter {
@@ -75,18 +75,18 @@ export class TLS implements ConnectionEncrypter {
     '@libp2p/connection-encryption'
   ]
 
-  async secureInbound <Stream extends SecurableStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureInbound <Stream extends MessageStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     return this._encrypt(conn, true, options)
   }
 
-  async secureOutbound <Stream extends SecurableStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async secureOutbound <Stream extends MessageStream = MultiaddrConnection> (conn: Stream, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     return this._encrypt(conn, false, options)
   }
 
   /**
    * Encrypt connection
    */
-  async _encrypt <Stream extends SecurableStream = MultiaddrConnection> (conn: Stream, isServer: boolean, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+  async _encrypt <Stream extends MessageStream = MultiaddrConnection> (conn: Stream, isServer: boolean, options?: SecureConnectionOptions): Promise<SecuredConnection> {
     const log = conn.log?.newScope('tls') ?? this.log
     let streamMuxer: StreamMuxerFactory | undefined
 
@@ -134,14 +134,14 @@ export class TLS implements ConnectionEncrypter {
     let socket: TLSSocket
 
     if (isServer) {
-      socket = new TLSSocket(itToStream(conn), {
+      socket = new TLSSocket(toNodeDuplex(conn), {
         ...opts,
         // require clients to send certificates
         requestCert: true
       })
     } else {
       socket = connect({
-        socket: itToStream(conn),
+        socket: toNodeDuplex(conn),
         ...opts
       })
     }
@@ -185,10 +185,7 @@ export class TLS implements ConnectionEncrypter {
 
             resolve({
               remotePeer,
-              conn: {
-                ...conn,
-                ...streamToIt(socket)
-              },
+              conn: toMessageStream(conn, socket),
               streamMuxer
             })
           })