Devp2p EventEmitter refactor

packages/client/src/net/peer/rlpxpeer.ts

@@ -137,11 +137,11 @@ export class RlpxPeer extends Peer {
       this.rlpxPeer = null
       this.connected = false
       this.config.events.emit(Event.PEER_DISCONNECTED, this)
-      this.rlpx?.removeListener('peer:error', peerErrorHandlerBound)
+      this.rlpx?.events.removeListener('peer:error', peerErrorHandlerBound)
     }
-    this.rlpx.on('peer:error', peerErrorHandlerBound)
-    this.rlpx.once('peer:added', peerAddedHandler.bind(this))
-    this.rlpx.once('peer:removed', peerRemovedHandler.bind(this))
+    this.rlpx.events.on('peer:error', peerErrorHandlerBound)
+    this.rlpx.events.once('peer:added', peerAddedHandler.bind(this))
+    this.rlpx.events.once('peer:removed', peerRemovedHandler.bind(this))
   }
 
   /**

packages/client/src/net/protocol/rlpxsender.ts

@@ -19,10 +19,10 @@ export class RlpxSender extends Sender {
     super()
 
     this.sender = rlpxProtocol
-    this.sender.on('status', (status: any) => {
+    this.sender.events.on('status', (status: any) => {
       this.status = status
     })
-    this.sender.on('message', (code: number, payload: any) => {
+    this.sender.events.on('message', (code: number, payload: any) => {
       this.emit('message', { code, payload })
     })
   }

packages/client/src/net/server/rlpxserver.ts

@@ -223,9 +223,9 @@ export class RlpxServer extends Server {
         dnsAddr: this.config.dnsAddr,
       })
 
-      this.dpt.on('error', (e: Error) => this.error(e))
+      this.dpt.events.on('error', (e: Error) => this.error(e))
 
-      this.dpt.on('listening', () => {
+      this.dpt.events.on('listening', () => {
         resolve()
       })
 
@@ -250,7 +250,7 @@ export class RlpxServer extends Server {
         common: this.config.chainCommon,
       })
 
-      this.rlpx.on('peer:added', async (rlpxPeer: Devp2pRLPxPeer) => {
+      this.rlpx.events.on('peer:added', async (rlpxPeer: Devp2pRLPxPeer) => {
         let peer: RlpxPeer | null = new RlpxPeer({
           config: this.config,
           id: bytesToUnprefixedHex(rlpxPeer.getId()!),
@@ -276,7 +276,7 @@ export class RlpxServer extends Server {
         }
       })
 
-      this.rlpx.on('peer:removed', (rlpxPeer: Devp2pRLPxPeer, reason: any) => {
+      this.rlpx.events.on('peer:removed', (rlpxPeer: Devp2pRLPxPeer, reason: any) => {
         const id = bytesToUnprefixedHex(rlpxPeer.getId() as Uint8Array)
         const peer = this.peers.get(id)
         if (peer) {
@@ -288,11 +288,13 @@ export class RlpxServer extends Server {
         }
       })
 
-      this.rlpx.on('peer:error', (rlpxPeer: Devp2pRLPxPeer, error: Error) => this.error(error))
+      this.rlpx.events.on('peer:error', (rlpxPeer: Devp2pRLPxPeer, error: Error) =>
+        this.error(error)
+      )
 
-      this.rlpx.on('error', (e: Error) => this.error(e))
+      this.rlpx.events.on('error', (e: Error) => this.error(e))
 
-      this.rlpx.on('listening', () => {
+      this.rlpx.events.on('listening', () => {
         this.config.events.emit(Event.SERVER_LISTENING, {
           transport: this.name,
           url: this.getRlpxInfo().enode ?? '',

packages/client/test/net/peer/rlpxpeer.spec.ts

@@ -1,12 +1,24 @@
-import { RLPx } from '@ethereumjs/devp2p'
 import { EventEmitter } from 'events'
 import { assert, describe, expect, it, vi } from 'vitest'
 
 import { Config } from '../../../src/config'
 import { Event } from '../../../src/types'
 
 describe('[RlpxPeer]', async () => {
-  vi.mock('@ethereumjs/devp2p')
+  vi.mock('@ethereumjs/devp2p', async () => {
+    const devp2p = await vi.importActual<any>('@ethereumjs/devp2p')
+    const RLPx = vi.fn().mockImplementation(() => {
+      return {
+        events: new EventEmitter(),
+        connect: vi.fn(),
+      }
+    })
+
+    return {
+      ...devp2p,
+      RLPx,
+    }
+  })
 
   const { RlpxPeer } = await import('../../../src/net/peer/rlpxpeer')
 
@@ -22,7 +34,7 @@ describe('[RlpxPeer]', async () => {
     assert.notOk(peer.connected, 'not connected')
   })
 
-  it('should compute capabilities', () => {
+  it('should compute capabilities', async () => {
     const protocols: any = [
       { name: 'eth', versions: [66] },
       { name: 'les', versions: [4] },
@@ -57,7 +69,7 @@ describe('[RlpxPeer]', async () => {
     proto0.open = vi.fn().mockResolvedValue(null)
     await peer.connect()
     assert.ok('connected successfully')
-    expect(RLPx.prototype.connect).toBeCalled()
+    expect(peer.rlpx!.connect).toBeCalled()
   })
 
   it('should handle peer events', async () => {
@@ -83,9 +95,9 @@ describe('[RlpxPeer]', async () => {
     peer.config.events.on(Event.PEER_DISCONNECTED, (rlpxPeer) =>
       assert.equal(rlpxPeer.pooled, false, 'got disconnected')
     )
-    peer.rlpx!.emit('peer:error', rlpxPeer, new Error('err0'))
-    peer.rlpx!.emit('peer:added', rlpxPeer)
-    peer.rlpx!.emit('peer:removed', rlpxPeer, 'reason')
+    peer.rlpx!.events.emit('peer:error', rlpxPeer, new Error('err0'))
+    peer.rlpx!.events.emit('peer:added', rlpxPeer)
+    peer.rlpx!.events.emit('peer:removed', rlpxPeer, 'reason')
     ;(peer as any).bindProtocols = vi.fn().mockRejectedValue(new Error('err1'))
     rlpxPeer.getDisconnectPrefix = vi.fn().mockImplementation((param: string) => {
       if (param === 'reason') throw new Error('err2')
@@ -98,8 +110,8 @@ describe('[RlpxPeer]', async () => {
       if (err.message === 'err1') assert.ok(true, 'got err1')
       if (err.message === 'err2') assert.ok(true, 'got err2')
     })
-    peer.rlpx!.emit('peer:added', rlpxPeer)
-    peer.rlpx!.emit('peer:removed', rlpxPeer, 'reason')
+    peer.rlpx!.events.emit('peer:added', rlpxPeer)
+    peer.rlpx!.events.emit('peer:removed', rlpxPeer, 'reason')
   })
 
   it('should accept peer connection', async () => {
@@ -114,8 +126,20 @@ describe('[RlpxPeer]', async () => {
   it('should bind protocols', async () => {
     const config = new Config({ transports: [], accountCache: 10000, storageCache: 1000 })
     const protocols = [{ name: 'proto0' }] as any
-    const peer = new RlpxPeer({ config, id: 'abcdef0123', protocols, host: '10.0.0.1', port: 1234 })
-    const proto0 = new (class Proto0 extends EventEmitter {})()
+    const peer = new RlpxPeer({
+      config,
+      id: 'abcdef0123',
+      protocols,
+      host: '10.0.0.1',
+      port: 1234,
+    })
+    class Proto0 {
+      events: EventEmitter
+      constructor() {
+        this.events = new EventEmitter()
+      }
+    }
+    const proto0 = new Proto0()
 
     const rlpxPeer = {
       getProtocols: vi.fn().mockReturnValue([proto0]),

packages/client/test/net/protocol/rlpxsender.spec.ts

@@ -27,27 +27,27 @@ describe('[RlpxSender]', () => {
   })
 
   it('should receive status', () => {
-    const rlpxProtocol = new EventEmitter()
+    const rlpxProtocol = { events: new EventEmitter() }
     const sender = new RlpxSender(rlpxProtocol as Devp2pETH)
     sender.on('status', (status: any) => {
       assert.equal(status.id, 5, 'status received')
       assert.equal(sender.status.id, 5, 'status getter')
     })
-    rlpxProtocol.emit('status', { id: 5 })
+    rlpxProtocol.events.emit('status', { id: 5 })
   })
 
   it('should receive message', () => {
-    const rlpxProtocol = new EventEmitter()
+    const rlpxProtocol = { events: new EventEmitter() }
     const sender = new RlpxSender(rlpxProtocol as Devp2pETH)
     sender.on('message', (message: any) => {
       assert.equal(message.code, 1, 'message received (code)')
       assert.equal(message.payload, 5, 'message received (payload)')
     })
-    rlpxProtocol.emit('message', 1, 5)
+    rlpxProtocol.events.emit('message', 1, 5)
   })
 
   it('should catch errors', () => {
-    const rlpxProtocol = new EventEmitter()
+    const rlpxProtocol = { events: new EventEmitter() }
     const sender = new RlpxSender(rlpxProtocol as Devp2pETH)
     assert.throws(() => sender.sendStatus({ id: 5 }), /not a function/, 'sendStatus error')
     assert.throws(() => sender.sendMessage(1, 5), /not a function/, 'sendMessage error')

packages/client/test/net/server/rlpxserver.spec.ts

@@ -72,7 +72,7 @@ describe('[RlpxServer]', async () => {
     server.rlpx = td.object()
     td.when(
       server.dpt!.bootstrap({ address: '10.0.0.1', udpPort: 1234, tcpPort: 1234 })
-    ).thenResolve()
+    ).thenResolve(undefined)
     td.when(
       (server.dpt! as any).bootstrap({ address: '10.0.0.2', udpPort: '1234', tcpPort: '1234' })
     ).thenReject(new Error('err0'))
@@ -237,7 +237,7 @@ describe('[RlpxServer]', async () => {
         assert.equal(err.message, 'err0', 'got error')
       })
     )
-    ;(server.dpt as any).emit('error', new Error('err0'))
+    server['dpt'].events.emit('error', new Error('err0'))
   })
 
   it('should init rlpx', async () => {
@@ -269,14 +269,14 @@ describe('[RlpxServer]', async () => {
         assert.deepEqual(info, { transport: 'rlpx', url: 'enode://ff@0.0.0.0:30303' }, 'listening')
       })
     )
-    server.rlpx!.emit('peer:added', rlpxPeer)
+    server.rlpx!.events.emit('peer:added', rlpxPeer)
     ;(server as any).peers.set('01', { id: '01' } as any)
-    server.rlpx!.emit('peer:removed', rlpxPeer)
-    server.rlpx!.emit('peer:error', rlpxPeer, new Error('err0'))
+    server.rlpx!.events.emit('peer:removed', rlpxPeer)
+    server.rlpx!.events.emit('peer:error', rlpxPeer, new Error('err0'))
 
     // @ts-ignore
     server.rlpx!.id = hexToBytes('0xff')
-    server.rlpx!.emit('listening')
+    server.rlpx!.events.emit('listening')
   })
 
   it('should handles errors from id-less peers', async () => {
@@ -293,7 +293,7 @@ describe('[RlpxServer]', async () => {
         assert.equal(err.message, 'err0', 'got error')
       })
     )
-    server.rlpx!.emit('peer:error', rlpxPeer, new Error('err0'))
+    server.rlpx!.events.emit('peer:error', rlpxPeer, new Error('err0'))
   })
 
   it('should reset td', () => {

packages/devp2p/examples/peer-communication-les.cts

@@ -48,7 +48,7 @@ const dpt = new devp2p.DPT(PRIVATE_KEY, {
 })
 
 /* eslint-disable no-console */
-dpt.on('error', (err) => console.error(chalk.red(`DPT error: ${err}`)))
+dpt.events.on('error', (err) => console.error(chalk.red(`DPT error: ${err}`)))
 
 /* eslint-disable @typescript-eslint/no-use-before-define */
 
@@ -61,9 +61,9 @@ const rlpx = new devp2p.RLPx(PRIVATE_KEY, {
   remoteClientIdFilter: REMOTE_CLIENTID_FILTER,
 })
 
-rlpx.on('error', (err) => console.error(chalk.red(`RLPx error: ${err.stack ?? err}`)))
+rlpx.events.on('error', (err) => console.error(chalk.red(`RLPx error: ${err.stack ?? err}`)))
 
-rlpx.on('peer:added', (peer) => {
+rlpx.events.on('peer:added', (peer) => {
   const addr = getPeerAddr(peer)
   const les = peer.getProtocols()[0]
   const requests: { headers: BlockHeader[]; bodies: any[] } = { headers: [], bodies: [] }
@@ -85,7 +85,7 @@ rlpx.on('peer:added', (peer) => {
     forkID: [hexToBytes('0x3b8e0691'), intToBytes(1)],
   })
 
-  les.once('status', (status: LES.Status) => {
+  les.events.once('status', (status: LES.Status) => {
     const msg = [
       Uint8Array.from([]),
       [
@@ -98,7 +98,7 @@ rlpx.on('peer:added', (peer) => {
     les.sendMessage(devp2p.LES.MESSAGE_CODES.GET_BLOCK_HEADERS, msg)
   })
 
-  les.on('message', async (code: LES.MESSAGE_CODES, payload: any) => {
+  les.events.on('message', async (code: LES.MESSAGE_CODES, payload: any) => {
     switch (code) {
       case devp2p.LES.MESSAGE_CODES.BLOCK_HEADERS: {
         if (payload[2].length > 1) {
@@ -148,7 +148,7 @@ rlpx.on('peer:added', (peer) => {
   })
 })
 
-rlpx.on('peer:removed', (peer, reasonCode, disconnectWe) => {
+rlpx.events.on('peer:removed', (peer, reasonCode, disconnectWe) => {
   const who = disconnectWe === true ? 'we disconnect' : 'peer disconnect'
   const total = rlpx.getPeers().length
   console.log(
@@ -160,7 +160,7 @@ rlpx.on('peer:removed', (peer, reasonCode, disconnectWe) => {
   )
 })
 
-rlpx.on('peer:error', (peer, err) => {
+rlpx.events.on('peer:error', (peer, err) => {
   if (err.code === 'ECONNRESET') return
 
   if (err instanceof Error) {