fix(emulator): remove event listeners in TCP/RTU transport stop() (#270)

Event listeners registered in start() were never cleaned up during stop(),
causing memory leaks in applications with repeated start/stop cycles.

TCP Transport:
- Each start() added 'initialized' and 'error' listeners that persisted
- Now stop() calls removeAllListeners() for both events before closing

RTU Transport:
- The 'error' listener added in start() persisted for server lifetime
- Now stop() calls removeAllListeners('error') before closing

Technical Implementation:
- Uses EventEmitter.prototype.removeAllListeners.call() to access the
  method since modbus-serial doesn't expose it in TypeScript declarations
- Added regression tests verifying listeners are cleaned up across cycles
- Improved test assertions with constants and explicit leak verification

Closes #253

Author: groupsky

packages/emulator/src/transports/rtu.test.ts

@@ -8,6 +8,10 @@ import { RtuTransport } from './rtu.js'
 
 // Store captured service vector for testing
 let capturedServiceVector: any = null
+let mockServerInstance: any
+let eventListeners: Map<string, Set<(...args: unknown[]) => void>>
+// Track removeAllListeners calls across all mock instances
+let allRemoveAllListenersCalls: Array<{ event?: string }> = []
 
 // Mock modbus-serial
 jest.mock('modbus-serial', () => {
@@ -17,17 +21,37 @@ jest.mock('modbus-serial', () => {
       .mockImplementation((vector: any, options: any, _serialportOptions?: any) => {
         // Capture service vector for testing
         capturedServiceVector = vector
+        eventListeners = new Map()
 
         // Call openCallback immediately to simulate successful connection
         if (options.openCallback) {
           setImmediate(() => options.openCallback(null))
         }
 
-        return {
+        mockServerInstance = {
           close: jest.fn((cb: (err: Error | null) => void) => cb(null)),
-          on: jest.fn(),
+          on: jest.fn((event: string, listener: (...args: unknown[]) => void) => {
+            // Track listeners
+            if (!eventListeners.has(event)) {
+              eventListeners.set(event, new Set())
+            }
+            eventListeners.get(event)!.add(listener)
+          }),
+          removeAllListeners: jest.fn((event?: string) => {
+            allRemoveAllListenersCalls.push({ event })
+            if (event) {
+              eventListeners.delete(event)
+            } else {
+              eventListeners.clear()
+            }
+          }),
+          listenerCount: jest.fn((event: string) => {
+            return eventListeners.get(event)?.size ?? 0
+          }),
           socks: new Map(),
         }
+
+        return mockServerInstance
       }),
   }
 })
@@ -37,6 +61,8 @@ describe('RtuTransport', () => {
 
   beforeEach(() => {
     capturedServiceVector = null
+    allRemoveAllListenersCalls = []
+    jest.clearAllMocks()
   })
 
   afterEach(async () => {
@@ -86,6 +112,40 @@ describe('RtuTransport', () => {
       await transport.start()
       await expect(transport.start()).rejects.toThrow('Transport already started')
     })
+
+    it('should not accumulate event listeners after repeated start/stop cycles', async () => {
+      // This test verifies the fix for issue #253: Event listener memory leak
+      // RTU transport has the same issue as TCP - 'error' listener is never removed
+
+      // Track listener counts across multiple transports
+      const listenerCounts: number[] = []
+
+      // Run 5 start/stop cycles
+      for (let i = 0; i < 5; i++) {
+        transport = new RtuTransport({ port: '/dev/ttyUSB0' })
+        await transport.start()
+
+        // Record listener count after start (only 'error' listener in RTU)
+        const count = mockServerInstance.listenerCount('error')
+        listenerCounts.push(count)
+
+        await transport.stop()
+      }
+
+      // After the fix: all cycles should have the same listener count (1 listener per cycle)
+      // Before the fix: listener count would grow with each cycle
+      const EXPECTED_LISTENERS_PER_CYCLE = 1 // only 'error' listener in RTU
+      const CYCLE_COUNT = 5
+
+      // Verify listener counts remain constant (primary regression test)
+      expect(listenerCounts).toEqual(Array(CYCLE_COUNT).fill(EXPECTED_LISTENERS_PER_CYCLE))
+
+      // Verify no listener accumulation - the most important behavior
+      const minCount = Math.min(...listenerCounts)
+      const maxCount = Math.max(...listenerCounts)
+      expect(maxCount).toBe(minCount) // All counts should be equal (no growth)
+      expect(maxCount).toBe(EXPECTED_LISTENERS_PER_CYCLE) // Should match expected count
+    })
   })
 
   describe('request/response handling', () => {
@@ -144,6 +204,7 @@ describe('RtuTransport', () => {
             closeCallback = cb
           }),
           on: jest.fn(),
+          removeAllListeners: jest.fn(),
           socks: new Map(),
         }
       })
@@ -173,6 +234,7 @@ describe('RtuTransport', () => {
               errorHandler = handler
             }
           }),
+          removeAllListeners: jest.fn(),
           socks: new Map(),
         }
       })

packages/emulator/src/transports/rtu.ts

@@ -4,6 +4,8 @@
  * This implementation uses modbus-serial ServerSerial for protocol handling.
  */
 
+import { EventEmitter } from 'events'
+
 import { ServerSerial } from 'modbus-serial'
 import type { IServiceVector } from 'modbus-serial/ServerTCP'
 
@@ -121,6 +123,10 @@ export class RtuTransport extends BaseTransport {
       return
     }
 
+    // Clean up event listeners to prevent memory leaks (issue #253)
+    // Use EventEmitter.prototype since modbus-serial doesn't expose these methods in types
+    EventEmitter.prototype.removeAllListeners.call(this.server, 'error')
+
     return new Promise<void>((resolve, reject) => {
       if (!this.server) {
         resolve()

packages/emulator/src/transports/tcp.test.ts

@@ -13,6 +13,9 @@ let capturedServiceVector: any = null
 let mockServerInstance: any
 let initializedListener: (() => void) | undefined
 let errorListener: ((err: Error) => void) | undefined
+let eventListeners: Map<string, Set<(...args: unknown[]) => void>>
+// Track removeAllListeners calls across all mock instances
+let allRemoveAllListenersCalls: Array<{ event?: string }> = []
 
 // Mock modbus-serial
 jest.mock('modbus-serial', () => {
@@ -24,6 +27,7 @@ jest.mock('modbus-serial', () => {
       // Reset listeners
       initializedListener = undefined
       errorListener = undefined
+      eventListeners = new Map()
 
       // Create mock instance
       mockServerInstance = {
@@ -34,6 +38,22 @@ jest.mock('modbus-serial', () => {
           } else if (event === 'error') {
             errorListener = listener
           }
+          // Track listeners
+          if (!eventListeners.has(event)) {
+            eventListeners.set(event, new Set())
+          }
+          eventListeners.get(event)!.add(listener)
+        }),
+        removeAllListeners: jest.fn((event?: string) => {
+          allRemoveAllListenersCalls.push({ event })
+          if (event) {
+            eventListeners.delete(event)
+          } else {
+            eventListeners.clear()
+          }
+        }),
+        listenerCount: jest.fn((event: string) => {
+          return eventListeners.get(event)?.size ?? 0
         }),
         socks: new Map(),
       }
@@ -55,6 +75,7 @@ describe('TcpTransport', () => {
 
   beforeEach(() => {
     capturedServiceVector = null
+    allRemoveAllListenersCalls = []
     jest.clearAllMocks()
   })
 
@@ -94,6 +115,43 @@ describe('TcpTransport', () => {
       await transport.start()
       await expect(transport.start()).rejects.toThrow('Transport already started')
     })
+
+    it('should not accumulate event listeners after repeated start/stop cycles', async () => {
+      // This test verifies the fix for issue #253: Event listener memory leak
+      // Before the fix, listeners are never removed during stop()
+      // Each start() adds listeners, so repeated cycles on the same server accumulate them
+
+      // Track listener counts across multiple transports using the same mock server
+      const listenerCounts: number[] = []
+
+      // Run 5 start/stop cycles
+      for (let i = 0; i < 5; i++) {
+        transport = new TcpTransport({ host: 'localhost', port: 502 })
+        await transport.start()
+
+        // Record listener count after start
+        const count =
+          mockServerInstance.listenerCount('initialized') +
+          mockServerInstance.listenerCount('error')
+        listenerCounts.push(count)
+
+        await transport.stop()
+      }
+
+      // After the fix: all cycles should have the same listener count (2 listeners per cycle)
+      // Before the fix: listener count would grow with each cycle
+      const EXPECTED_LISTENERS_PER_CYCLE = 2 // 'initialized' + 'error'
+      const CYCLE_COUNT = 5
+
+      // Verify listener counts remain constant (primary regression test)
+      expect(listenerCounts).toEqual(Array(CYCLE_COUNT).fill(EXPECTED_LISTENERS_PER_CYCLE))
+
+      // Verify no listener accumulation - the most important behavior
+      const minCount = Math.min(...listenerCounts)
+      const maxCount = Math.max(...listenerCounts)
+      expect(maxCount).toBe(minCount) // All counts should be equal (no growth)
+      expect(maxCount).toBe(EXPECTED_LISTENERS_PER_CYCLE) // Should match expected count
+    })
   })
 
   describe('request/response handling', () => {
@@ -140,6 +198,7 @@ describe('TcpTransport', () => {
               errorListener = listener
             }
           }),
+          removeAllListeners: jest.fn(),
           socks: new Map(),
         }
 

packages/emulator/src/transports/tcp.ts

@@ -4,6 +4,8 @@
  * This implementation uses modbus-serial ServerTCP for protocol handling.
  */
 
+import { EventEmitter } from 'events'
+
 import { ServerTCP } from 'modbus-serial'
 import type { IServiceVector } from 'modbus-serial/ServerTCP'
 
@@ -101,6 +103,11 @@ export class TcpTransport extends BaseTransport {
       return
     }
 
+    // Clean up event listeners to prevent memory leaks (issue #253)
+    // Use EventEmitter.prototype since modbus-serial doesn't expose these methods in types
+    EventEmitter.prototype.removeAllListeners.call(this.server, 'initialized')
+    EventEmitter.prototype.removeAllListeners.call(this.server, 'error')
+
     return new Promise<void>((resolve, reject) => {
       if (!this.server) {
         resolve()