cleanup

Author: sfast

scripts/zmq-monitor/dealer-example.js

@@ -0,0 +1,70 @@
+/* eslint-disable no-console */
+const zmq = require('zeromq');
+
+(async () => {
+  const address = process.env.ADDR || 'tcp://127.0.0.1:7777';
+  
+  // Simple config - ZeroMQ handles reconnection natively
+  const dealer = new zmq.Dealer({ 
+    routingId: 'dealer-monitor',
+    reconnectInterval: 5000,      // Retry every 5s (or -1 to disable)
+    reconnectMaxInterval: 20000   // Max backoff 20s (0 = constant interval)
+  });
+
+  console.log('[dealer] Config:', { address, ...dealer });
+  console.log('[dealer] Starting...\n');
+
+  // Attach event listeners (just for observability)
+  const on = (name) => dealer.events.on(name, (...args) => {
+    const ts = new Date().toISOString();
+    console.log(`[${ts}] 📡 ${name}`, ...args);
+  });
+  
+  ['connect', 'disconnect', 'connect_retry', 'connect:retry'].forEach(on);
+
+  await dealer.connect(address);
+  console.log(`[dealer] ✅ Connect issued to ${address}\n`);
+
+  // Track connection state for smart sending
+  let isConnected = false;
+  dealer.events.on('connect', () => {
+    isConnected = true;
+    console.log('[dealer] 🟢 CONNECTED - messages will be sent\n');
+  });
+  dealer.events.on('disconnect', () => {
+    isConnected = false;
+    console.log('[dealer] 🔴 DISCONNECTED - messages will be queued\n');
+  });
+
+  // Send periodic pings (only when connected)
+  let i = 0;
+  setInterval(async () => {
+    if (!isConnected) {
+      console.log(`[dealer] ⏸️  Skip ping-${i++} (offline)`);
+      return;
+    }
+    
+    try {
+      const msg = `ping-${i++}`;
+      await dealer.send(Buffer.from(msg));
+      console.log(`[dealer] 📤 Sent: ${msg}`);
+    } catch (e) {
+      console.log(`[dealer] ❌ Send error: ${e.message}`);
+    }
+  }, 2000);
+
+  // Receive echoes
+  (async () => {
+    for await (const [msg] of dealer) {
+      console.log(`[dealer] 📥 Received: ${msg.toString()}\n`);
+    }
+  })();
+
+  // Cleanup
+  process.on('SIGINT', () => {
+    console.log('\n[dealer] 🛑 Shutting down...');
+    dealer.close();
+    process.exit(0);
+  });
+})();
+

scripts/zmq-monitor/router-example.js

@@ -0,0 +1,83 @@
+/* eslint-disable no-console */
+const zmq = require('zeromq');
+
+(async () => {
+  const address = process.env.ADDR || 'tcp://127.0.0.1:7778';
+  const sock = new zmq.Router({ routingId: 'router-monitor-7778' });
+
+  // Monitor common router-side events
+  const on = (name) => sock.events.on(name, (...args) => {
+    const ts = new Date().toISOString();
+    console.log(`[${ts}] ROUTER events -> ${name}`, ...args);
+  });
+  [
+    'listening',
+    'accept',
+    'accept_error',
+    'bind_error',
+    'closed',
+    'close_error',
+    'disconnect',
+  ].forEach(on);
+
+  await sock.bind(address);
+  console.log(`[router] bound to ${address}`);
+
+  // Manual control: you can unbind/rebind the router yourself from the terminal
+  // Example:
+  //   ADDR=tcp://127.0.0.1:7777 node scripts/zmq-monitor/router.js
+  // Then in another terminal, stop and restart this process to simulate downtime
+
+  // Track connected dealers
+  const dealers = new Map(); // routingId -> { lastSeen, messageCount }
+
+  console.log('[router] Waiting for messages...');
+
+  // Echo replies back to sender and track stats
+  for await (const msg of sock) {
+    console.log('[router] 🔔 Received message with', msg.length, 'frames');
+    
+    // Router receives: [routingId, data] (2 frames)
+    // ZeroMQ handles the empty delimiter internally in v6+
+    try {
+      if (msg.length >= 2) {
+        const [routingId, ...dataFrames] = msg;
+        const routingIdStr = routingId.toString();
+        const data = dataFrames[0]?.toString() || '';
+        
+        console.log(`[router] 🆔 Routing ID: ${routingIdStr}`);
+        console.log(`[router] 📦 Data: ${data}`);
+        
+        // Track dealer stats
+        if (!dealers.has(routingIdStr)) {
+          dealers.set(routingIdStr, { lastSeen: Date.now(), messageCount: 0 });
+          console.log(`[router] 🔗 New dealer connected: ${routingIdStr}`);
+        }
+        
+        const dealer = dealers.get(routingIdStr);
+        dealer.lastSeen = Date.now();
+        dealer.messageCount++;
+        
+        // Log received message
+        if (data.startsWith('ping-')) {
+          console.log(`[router] 📩 Received from ${routingIdStr}: ${data} (total: ${dealer.messageCount})`);
+        } else {
+          console.log(`[router] 📩 Received from ${routingIdStr}: ${data}`);
+        }
+        
+        // Echo back (router sends: [routingId, data])
+        await sock.send([routingId, ...dataFrames]);
+        console.log(`[router] 📤 Echoed back to ${routingIdStr}`);
+      } else {
+        console.log(`[router] ⚠️  Unexpected frame count: ${msg.length}`);
+      }
+    } catch (e) {
+      console.log('[router] ❌ Error:', e && e.message);
+      console.log('[router] Stack:', e && e.stack);
+    }
+  }
+  
+  console.log('[router] Message loop ended');
+})(); 
+
+

src/protocol/client.js

@@ -236,12 +236,11 @@ export default class Client extends Protocol {
     const socket = this._getSocket()
 
     try {
-      // Connect transport
-      await socket.connect(serverAddress, timeout)
-
-      _scope.serverPeerInfo.setState('CONNECTED')
+      // Issue connect (non-blocking - returns immediately)
+      // ZeroMQ will connect in background and emit TRANSPORT_READY when ready
+      await socket.connect(serverAddress)
 
-      // Wait for handshake to complete (CLIENT_READY event)
+      // Wait for handshake to complete (includes implicit transport ready wait)
       await new Promise((resolve, reject) => {
         // Use provided timeout, then config, then global default, finally 10s
         const config = this.getConfig()
@@ -268,18 +267,17 @@ export default class Client extends Protocol {
 
     this._stopPing()
 
-    // Notify server
-    if (this.isReady()) {
-      try {
-        // ✅ Use internal API to send system event (client stop)
-        this._sendSystemTick({
-          event: ProtocolSystemEvent.CLIENT_STOP,
-          data: { clientId: this.getId() }
-        })
-      } catch (err) {
-        this.debugMode() && this.logger?.error('Error sending client stop: ', err)
-      }
+    // Try to notify server
+    try {
+      // ✅ Use internal API to send system event (client stop)
+      this._sendSystemTick({
+        event: ProtocolSystemEvent.CLIENT_STOP,
+        data: { clientId: this.getId() }
+      })
+    } catch (err) {
+      this.debugMode() && this.logger?.error('Error sending client stop: ', err)
     }
+    
 
     // disconnect from transport and detach listeners
     await super.disconnect();

src/protocol/protocol.js

@@ -808,20 +808,32 @@ export default class Protocol extends EventEmitter {
    */
   async disconnect () {
     let { socket } = _private.get(this)
-    this._detachSocketEventHandlers(socket)
-
     await socket.disconnect();
   }
 
+  /**
+   * Unbind protocol from transport events without closing or rejecting pending.
+   * - Idempotent: safe to call multiple times
+   * - Does NOT set closed flag
+   * - Does NOT reject pending requests
+   * - Does NOT close underlying transport
+   */
+
+  async unbind () {
+    let { socket } = _private.get(this)
+    // Keep socket event handlers attached so further transport events (e.g., CLOSED)
+    // still propagate through Protocol to consumers. Just unbind transport here.
+    await socket.unbind();
+  }
+
   /**
    * Close the protocol and cleanup resources.
    * - Idempotent
    * - Detaches protocol-attached socket listeners
    * - Rejects and clears pending requests
    * - Optionally closes the underlying transport
    * 
-   * @param {Object} [options]
-   * @param {boolean} [options.closeTransport=true] - Whether to close the socket
+   * @param {boolean} [closeTransport=false] - Whether to close the socket
    */
   async close (closeTransport = false) {
     let _scope = _private.get(this)

src/protocol/server.js

@@ -190,29 +190,26 @@ export default class Server extends Protocol {
   async unbind () {
     this._stopHealthChecks()
 
-    // Notify all clients
-    if (this.isOnline()) {
-        try {
-          this.tick({
-            event: ProtocolSystemEvent.SERVER_STOP,
-            data: { serverId: this.getId() }
-          })
-    } catch (err) {
-        // Ignore if offline
+    // Notify all clients individually with system event before unbind
+    try {
+      let { clientPeers } = _private.get(this)
+      for (const clientId of clientPeers.keys()) {
+        this._sendSystemTick({
+          to: clientId,
+          event: ProtocolSystemEvent.SERVER_STOP,
+          data: { serverId: this.getId() }
+        })
       }
+    } catch (err) {
+      this.debugMode() && this.logger?.error('Error sending server stop: ', err)
     }
 
-    // ✅ Use Protocol's socket (via protected method)
-    const socket = this._getSocket()
-    await socket.unbind()
+    await super.unbind()
   }
 
   async close () {
     await this.unbind()
-    
-    // ✅ Use Protocol's socket (via protected method)
-    const socket = this._getSocket()
-    await socket.close()
+    await super.close(true)
   }
 
   getAddress () {

src/transport/errors.js

@@ -31,7 +31,7 @@ export const TransportErrorCode = {
 
   // Address errors
   INVALID_ADDRESS: 'TRANSPORT_INVALID_ADDRESS',           // Invalid address format
-  ADDRESS_REQUIRED: 'TRANSPORT_ADDRESS_REQUIRED',         // Address not provided
+  // ADDRESS_REQUIRED: 'TRANSPORT_ADDRESS_REQUIRED',         // Address not provided
 
   // Lifecycle errors
   CLOSE_FAILED: 'TRANSPORT_CLOSE_FAILED'                  // Failed to close cleanly

src/transport/zeromq/config.js

@@ -108,6 +108,7 @@ export const ZMQConfigDefaults = {
    * ZMQ_RECONNECT_IVL: Initial reconnection interval in milliseconds
    * How often ZeroMQ tries to reconnect after losing connection
    * Default: 100ms (fast reconnection)
+   * -1 means no reconnection
    */
   ZMQ_RECONNECT_IVL: 100,