import * as zmq from 'zeromq'; import {Mokka} from '../consensus/main'; class ZMQMokka extends Mokka { private sockets: Map = new Map(); public async initialize() { this.logger.info(`initializing reply socket on port ${this.address}`); const socket = new zmq.Pair({receiveHighWaterMark: 10}); await socket.bind(this.address); this.sockets.set(this.address, socket); for await (const [msg] of this.sockets.get(this.address)) { await this.emitPacket(msg); } } /** * The message to write. * * @param {string} address The peer address * @param {Object} packet The packet to write to the connection. * @api private */ public async write(address: string, packet: Buffer): Promise { if (!this.sockets.has(address)) { const socket = new zmq.Pair({receiveHighWaterMark: 10}); socket.connect(address); this.sockets.set(address, socket); } try { await this.sockets.get(address).send(packet); } catch (e) { this.sockets.get(address).close(); const socket = new zmq.Pair({receiveHighWaterMark: 10}); socket.connect(address); this.sockets.set(address, socket); } } public async disconnect(): Promise { await super.disconnect(); for (const socket of this.sockets.values()) { socket.close(); } } public async connect(): Promise { this.initialize(); await super.connect(); } } export default ZMQMokka;