Class XSubscriber

constructor

• new XSubscriber(options?): XSubscriber

• Parameters

  • Optionaloptions: {
        affinity?: number;
        backlog?: number;
        connectTimeout?: number;
        context?: Context;
        curvePublicKey?: null | string;
        curveSecretKey?: null | string;
        curveServer?: boolean;
        curveServerKey?: null | string;
        gssapiPlainText?: boolean;
        gssapiPrincipal?: null | string;
        gssapiPrincipalNameType?: "hostBased" | "userName" | "krb5Principal";
        gssapiServer?: boolean;
        gssapiServicePrincipal?: null | string;
        gssapiServicePrincipalNameType?: "hostBased" | "userName" | "krb5Principal";
        handshakeInterval?: number;
        heartbeatInterval?: number;
        heartbeatTimeout?: number;
        heartbeatTimeToLive?: number;
        immediate?: boolean;
        interface?: null | string;
        ipv6?: boolean;
        linger?: number;
        loopbackFastPath?: boolean;
        maxMessageSize?: number;
        multicastHops?: number;
        multicastMaxTransportDataUnit?: number;
        plainPassword?: null | string;
        plainServer?: boolean;
        plainUsername?: null | string;
        rate?: number;
        receiveBufferSize?: number;
        receiveHighWaterMark?: number;
        receiveTimeout?: number;
        reconnectInterval?: number;
        reconnectMaxInterval?: number;
        recoveryInterval?: number;
        sendBufferSize?: number;
        sendHighWaterMark?: number;
        sendTimeout?: number;
        socksProxy?: null | string;
        tcpAcceptFilter?: null | string;
        tcpKeepalive?: number;
        tcpKeepaliveCount?: number;
        tcpKeepaliveIdle?: number;
        tcpKeepaliveInterval?: number;
        tcpMaxRetransmitTimeout?: number;
        typeOfService?: number;
        vmciBufferMaxSize?: number;
        vmciBufferMinSize?: number;
        vmciBufferSize?: number;
        vmciConnectTimeout?: number;
        zapDomain?: null | string;
        zapEnforceDomain?: boolean;
    }

    • Optionalaffinity?: number

      ZMQ_AFFINITY

      I/O thread affinity, which determines which threads from the ØMQ I/O thread pool associated with the socket's context shall handle newly created connections.

      Note: This value is a bit mask, but values higher than Number.MAX_SAFE_INTEGER may not be represented accurately! This currently means that configurations beyond 52 threads are unreliable.

    • Optionalbacklog?: number

      ZMQ_BACKLOG

      Maximum length of the queue of outstanding peer connections for the specified socket. This only applies to connection-oriented transports.

    • OptionalconnectTimeout?: number

      ZMQ_CONNECT_TIMEOUT

      Sets how long to wait before timing-out a connect() system call. The connect() system call normally takes a long time before it returns a time out error. Setting this option allows the library to time out the call at an earlier interval.

    • Optionalcontext?: Context

    • OptionalcurvePublicKey?: null | string

      ZMQ_CURVE_PUBLICKEY

      Sets the socket's long term public key. You must set this on CURVE client sockets. A server socket does not need to know its own public key. You can create a new keypair with curveKeyPair().

    • OptionalcurveSecretKey?: null | string

      ZMQ_CURVE_SECRETKEY

      Sets the socket's long term secret key. You must set this on both CURVE client and server sockets. You can create a new keypair with curveKeyPair().

    • OptionalcurveServer?: boolean

      ZMQ_CURVE_SERVER

      Defines whether the socket will act as server for CURVE security. A value of true means the socket will act as CURVE server. A value of false means the socket will not act as CURVE server, and its security role then depends on other option settings.

    • OptionalcurveServerKey?: null | string

      ZMQ_CURVE_SERVERKEY

      Sets the socket's long term server key. This is the public key of the CURVE server socket. You must set this on CURVE client sockets. This key must have been generated together with the server's secret key. You can create a new keypair with curveKeyPair().

    • OptionalgssapiPlainText?: boolean

    • OptionalgssapiPrincipal?: null | string

    • OptionalgssapiPrincipalNameType?: "hostBased" | "userName" | "krb5Principal"

    • OptionalgssapiServer?: boolean

    • OptionalgssapiServicePrincipal?: null | string

    • OptionalgssapiServicePrincipalNameType?: "hostBased" | "userName" | "krb5Principal"

    • OptionalhandshakeInterval?: number

      ZMQ_HANDSHAKE_IVL

      Handshaking is the exchange of socket configuration information (socket type, identity, security) that occurs when a connection is first opened (only for connection-oriented transports). If handshaking does not complete within the configured time, the connection shall be closed. The value 0 means no handshake time limit.

    • OptionalheartbeatInterval?: number

      ZMQ_HEARTBEAT_IVL

      Interval in milliseconds between sending ZMTP heartbeats for the specified socket. If this option is greater than 0, then a PING ZMTP command will be sent after every interval.

    • OptionalheartbeatTimeout?: number

      ZMQ_HEARTBEAT_TIMEOUT

      How long (in milliseconds) to wait before timing-out a connection after sending a PING ZMTP command and not receiving any traffic. This option is only valid if heartbeatInterval is greater than 0. The connection will time out if there is no traffic received after sending the PING command. The received traffic does not have to be a PONG command - any received traffic will cancel the timeout.

    • OptionalheartbeatTimeToLive?: number

      ZMQ_HEARTBEAT_TTL

      The timeout in milliseconds on the remote peer for ZMTP heartbeats. If this option is greater than 0, the remote side shall time out the connection if it does not receive any more traffic within the TTL period. This option does not have any effect if heartbeatInterval is 0. Internally, this value is rounded down to the nearest decisecond, any value less than 100 will have no effect.

    • Optionalimmediate?: boolean

      ZMQ_IMMEDIATE

      By default queues will fill on outgoing connections even if the connection has not completed. This can lead to "lost" messages on sockets with round-robin routing (Request, Push, Dealer). If this option is set to true, messages shall be queued only to completed connections. This will cause the socket to block if there are no other connections, but will prevent queues from filling on pipes awaiting connection.

    • Optionalinterface?: null | string

      ZMQ_BINDTODEVICE

      Binds the socket to the given network interface (Linux only). Allows to use Linux VRF, see: https://www.kernel.org/doc/Documentation/networking/vrf.txt. Requires the program to be ran as root or with CAP_NET_RAW.

    • Optionalipv6?: boolean

      ZMQ_IPV6

      Enable or disable IPv6. When IPv6 is enabled, the socket will connect to, or accept connections from, both IPv4 and IPv6 hosts.

    • Optionallinger?: number

      ZMQ_LINGER

      Determines how long pending messages which have yet to be sent to a peer shall linger in memory after a socket is closed with close().

    • OptionalloopbackFastPath?: boolean

      ZMQ_LOOPBACK_FASTPATH

      Enable faster TCP connections on loopback devices. An application can enable this option to reduce the latency and improve the performance of loopback operations on a TCP socket on Windows.

      Windows

    • OptionalmaxMessageSize?: number

      ZMQ_MAXMSGSIZE

      Limits the size of the inbound message. If a peer sends a message larger than the limit it is disconnected. Value of -1 means no limit.

    • OptionalmulticastHops?: number

      ZMQ_MULTICAST_HOPS

      Sets the time-to-live field in every multicast packet sent from this socket. The default is 1 which means that the multicast packets don't leave the local network.

    • OptionalmulticastMaxTransportDataUnit?: number

      ZMQ_MULTICAST_MAXTPDU

      Sets the maximum transport data unit size used for outbound multicast packets. This must be set at or below the minimum Maximum Transmission Unit (MTU) for all network paths over which multicast reception is required.

    • OptionalplainPassword?: null | string

      ZMQ_PLAIN_PASSWORD

      Sets the password for outgoing connections over TCP or IPC. If you set this to a non-null value, the security mechanism used for connections shall be PLAIN.

    • OptionalplainServer?: boolean

      ZMQ_PLAIN_SERVER

      Defines whether the socket will act as server for PLAIN security. A value of true means the socket will act as PLAIN server. A value of false means the socket will not act as PLAIN server, and its security role then depends on other option settings.

    • OptionalplainUsername?: null | string

      ZMQ_PLAIN_USERNAME

      Sets the username for outgoing connections over TCP or IPC. If you set this to a non-null value, the security mechanism used for connections shall be PLAIN.

    • Optionalrate?: number

      ZMQ_RATE

      Maximum send or receive data rate for multicast transports such as pgm.

    • OptionalreceiveBufferSize?: number

      ZMQ_RCVBUF

      Underlying kernel receive buffer size in bytes. A value of -1 means leave the OS default unchanged.

    • OptionalreceiveHighWaterMark?: number

      ZMQ_RCVHWM

      The high water mark is a hard limit on the maximum number of incoming messages ØMQ shall queue in memory for any single peer that the specified socket is communicating with. A value of zero means no limit.

      If this limit has been reached the socket shall enter an exceptional state and depending on the socket type, ØMQ shall take appropriate action such as blocking or dropping sent messages.

    • OptionalreceiveTimeout?: number

      ZMQ_RCVTIMEO

      Sets the timeout receiving messages on the socket. If the value is 0, receive() will return a rejected promise immediately if there is no message to receive. If the value is -1, it will wait asynchronously until a message is available. For all other values, it will wait for a message for that amount of time before rejecting.

    • OptionalreconnectInterval?: number

      ZMQ_RECONNECT_IVL

      Period ØMQ shall wait between attempts to reconnect disconnected peers when using connection-oriented transports. The value -1 means no reconnection.

    • OptionalreconnectMaxInterval?: number

      ZMQ_RECONNECT_IVL_MAX

      Maximum period ØMQ shall wait between attempts to reconnect. On each reconnect attempt, the previous interval shall be doubled until reconnectMaxInterval is reached. This allows for exponential backoff strategy. Zero (the default) means no exponential backoff is performed and reconnect interval calculations are only based on reconnectInterval.

    • OptionalrecoveryInterval?: number

      ZMQ_RECOVERY_IVL

      Maximum time in milliseconds that a receiver can be absent from a multicast group before unrecoverable data loss will occur.

    • OptionalsendBufferSize?: number

      ZMQ_SNDBUF

      Underlying kernel transmit buffer size in bytes. A value of -1 means leave the OS default unchanged.

    • OptionalsendHighWaterMark?: number

      ZMQ_SNDHWM

      The high water mark is a hard limit on the maximum number of outgoing messages ØMQ shall queue in memory for any single peer that the specified socket is communicating with. A value of zero means no limit.

      If this limit has been reached the socket shall enter an exceptional state and depending on the socket type, ØMQ shall take appropriate action such as blocking or dropping sent messages.

    • OptionalsendTimeout?: number

      ZMQ_SNDTIMEO

      Sets the timeout for sending messages on the socket. If the value is 0, send() will return a rejected promise immediately if the message cannot be sent. If the value is -1, it will wait asynchronously until the message is sent. For all other values, it will try to send the message for that amount of time before rejecting.

    • OptionalsocksProxy?: null | string

      ZMQ_SOCKS_PROXY

      The SOCKS5 proxy address that shall be used by the socket for the TCP connection(s). Does not support SOCKS5 authentication. If the endpoints are domain names instead of addresses they shall not be resolved and they shall be forwarded unchanged to the SOCKS proxy service in the client connection request message (address type 0x03 domain name).

    • OptionaltcpAcceptFilter?: null | string

      ZMQ_TCP_ACCEPT_FILTER

      Assign a filter that will be applied for each new TCP transport connection on a listening socket. If no filters are applied, then the TCP transport allows connections from any IP address. If at least one filter is applied then new connection source IP should be matched. To clear all filters set to null. Filter is a string with IPv6 or IPv4 CIDR.

    • OptionaltcpKeepalive?: number

      ZMQ_TCP_KEEPALIVE

      Override SO_KEEPALIVE socket option (if supported by OS). The default value of -1 leaves it to the OS default.

    • OptionaltcpKeepaliveCount?: number

      ZMQ_TCP_KEEPALIVE_CNT

      Overrides TCP_KEEPCNT socket option (if supported by OS). The default value of -1 leaves it to the OS default.

    • OptionaltcpKeepaliveIdle?: number

      ZMQ_TCP_KEEPALIVE_IDLE

      Overrides TCP_KEEPIDLE / TCP_KEEPALIVE socket option (if supported by OS). The default value of -1 leaves it to the OS default.

    • OptionaltcpKeepaliveInterval?: number

      ZMQ_TCP_KEEPALIVE_INTVL

      Overrides TCP_KEEPINTVL socket option (if supported by the OS). The default value of -1 leaves it to the OS default.

    • OptionaltcpMaxRetransmitTimeout?: number

      ZMQ_TCP_MAXRT

      Sets how long before an unacknowledged TCP retransmit times out (if supported by the OS). The system normally attempts many TCP retransmits following an exponential backoff strategy. This means that after a network outage, it may take a long time before the session can be re-established. Setting this option allows the timeout to happen at a shorter interval.

    • OptionaltypeOfService?: number

      ZMQ_TOS

      Sets the ToS fields (the Differentiated Services (DS) and Explicit Congestion Notification (ECN) field) of the IP header. The ToS field is typically used to specify a packet's priority. The availability of this option is dependent on intermediate network equipment that inspect the ToS field and provide a path for low-delay, high-throughput, highly-reliable service, etc.

    • OptionalvmciBufferMaxSize?: number

      ZMQ_VMCI_BUFFER_MAX_SIZE

      Maximum size of the underlying buffer for the socket. Used during negotiation before the connection is established. For vmci:// transports only.

    • OptionalvmciBufferMinSize?: number

      ZMQ_VMCI_BUFFER_MIN_SIZE

      Minimum size of the underlying buffer for the socket. Used during negotiation before the connection is established. For vmci:// transports only.

    • OptionalvmciBufferSize?: number

      ZMQ_VMCI_BUFFER_SIZE

      The size of the underlying buffer for the socket. Used during negotiation before the connection is established. For vmci:// transports only.

    • OptionalvmciConnectTimeout?: number

      ZMQ_VMCI_CONNECT_TIMEOUT

      Connection timeout for the socket. For vmci:// transports only.

    • OptionalzapDomain?: null | string

      ZMQ_ZAP_DOMAIN

      Sets the domain for ZAP (ZMQ RFC 27) authentication. For NULL security (the default on all tcp:// connections), ZAP authentication only happens if you set a non-empty domain. For PLAIN and CURVE security, ZAP requests are always made, if there is a ZAP handler present. See http://rfc.zeromq.org/spec:27 for more details.

    • OptionalzapEnforceDomain?: boolean

      ZMQ_ZAP_ENFORCE_DOMAIN

      The ZAP (ZMQ RFC 27) authentication protocol specifies that a domain must always be set. Older versions of libzmq did not follow the spec and allowed an empty domain to be set. This option can be used to enabled or disable the stricter, backward incompatible behaviour. For now it is disabled by default, but in a future version it will be enabled by default.

  Returns XSubscriber

multicastHops

receiveBufferSize

receiveHighWaterMark

receiveTimeout

sendBufferSize

sendHighWaterMark

sendTimeout

[asyncIterator]

• [asyncIterator](): AsyncIterator<Buffer[], undefined, undefined>

• Asynchronously iterate over messages becoming available on the socket. When the socket is closed with Socket.close(), the iterator will return. Returning early from the iterator will not close the socket unless it also goes out of scope.

  for await (const [msg] of socket) {
    // handle messages
  }
  Copy

  Returns AsyncIterator<Buffer[], undefined, undefined>

receive

• receive(): Promise<Buffer[]>

• Waits for the next single or multipart message to become availeble on the socket. Reads a message immediately if possible. If no messages can be read, it will wait asynchonously. The promise will be resolved with an array containing the parts of the next message when available.

  const [msg] = await socket.receive()
  const [part1, part2] = await socket.receive()
  Copy

  Reading may fail (eventually) if the socket has been configured with a receiveTimeout.

  A call to receive() is guaranteed to return with a resolved promise immediately if a message could be read from the socket directly.

  Only one asynchronously blocking call to receive() can be in progress simultaneously. If you call receive() again on the same socket it will return a rejected promise with an EBUSY error. For example, if no messages can be read and no await is used:

  socket.receive() // -> pending promise until read is possible
  socket.receive() // -> promise rejection with `EBUSY` error
  Copy

  Note: Due to the nature of Node.js and to avoid blocking the main thread, this method always attempts to read messages with the ZMQ_DONTWAIT flag. It polls asynchronously if reading is not currently possible. This means that all functionality related to timeouts and blocking behaviour is reimplemented in the Node.js bindings. Any differences in behaviour with the native ZMQ library is considered a bug.

  Returns Promise<Buffer[]>

  Resolved with message parts that were successfully read.

send

• send(message, ...options): Promise<void>

• Sends a single message or a multipart message on the socket. Queues the message immediately if possible, and returns a resolved promise. If the message cannot be queued because the high water mark has been reached, it will wait asynchronously. The promise will be resolved when the message was queued successfully.

  await socket.send("hello world")
  await socket.send(["hello", "world"])
  Copy

  Queueing may fail eventually if the socket has been configured with a sendTimeout.

  A call to send() is guaranteed to return with a resolved promise immediately if the message could be queued directly.

  Only one asynchronously blocking call to send() may be executed simultaneously. If you call send() again on a socket that is in the mute state it will return a rejected promise with an EBUSY error.

  The reason for disallowing multiple send() calls simultaneously is that it could create an implicit queue of unsendable outgoing messages. This would circumvent the socket's sendHighWaterMark. Such an implementation could even exhaust all system memory and cause the Node.js process to abort.

  For most application you should not notice this implementation detail. Only in rare occasions will a call to send() that does not resolve immediately be undesired. Here are some common scenarios:

  • If you wish to send a message, use await send(...). ZeroMQ socket types have been carefully designed to give you the correct blocking behaviour on the chosen socket type in almost all cases:

    • If sending is not possible, it is often better to wait than to continue as if nothing happened. For example, on a Request socket, you can only receive a reply once a message has been sent; so waiting until a message could be queued before continuing with the rest of the program (likely to read from the socket) is required.

    • Certain socket types (such as Router) will always allow queueing messages and await send(...) won't delay any code that comes after. This makes sense for routers, since typically you don't want a single send operation to stop the handling of other incoming or outgoing messages.

  • If you wish to send on an occasionally blocking socket (for example on a Router with the Router.mandatory option set, or on a Dealer) and you're 100% certain that dropping a message is better than blocking, then you can set the sendTimeout option to 0 to effectively force send() to always resolve immediately. Be prepared to catch exceptions if sending a message is not immediately possible.

  • If you wish to send on a socket and messages should be queued before they are dropped, you should implement a simple queue in JavaScript. Such a queue is not provided by this library because most real world applications need to deal with undeliverable messages in more complex ways - for example, they might need to reply with a status message; or first retry delivery a certain number of times before giving up.

  Parameters

  • message: MessageLike | MessageLike[]

    Single message or multipart message to queue for sending.
  • Rest...options: []

    Any options, if applicable to the socket type (DRAFT only).

  Returns Promise<void>

  Resolved when the message was successfully queued.