SMQ Cluster Management

Clustering can be used for the following:

• Remove the single point of failure if the one and only broker fails.
• Scale the solution horizontally to meet high demand, high connection counts, or both.
• Bridge two or more Barracuda App Servers together and use SMQ for the communication -- e.g. Bypassing SOP

Note:

When using an SMQ cluster for scaling, remember that cross-node message delivery adds overhead. If publishers and subscribers for the same topics are spread across multiple nodes, the cluster bus must relay more traffic between brokers. To minimize this load, group clients that frequently communicate on the same node, typically using a load balancer or regional affinity strategy. This keeps most traffic local and improves overall cluster efficiency.

Download the ready-to-run SMQ cluster example from GitHub, which is designed to run on your own computer.

Enabling clustering is best illustrated with an example:

smq=require"smq.hub".create()
cluster=require"smq.cluster".create(smq,"cluster password")
conn,err=require"smq.conn".create(cluster, 1900)
if conn then
   --Cluster nodes. Use names if you have DNS
   local list={"192.168.1.100","192.168.1.101", "192.168.1.102"}
   conn:setlist(list) -- Connect the nodes
end

Figure 1: MTL implicitly created by Cluster Manager

• Line 1: Create an SMQ broker.
• Line 2: Attach a Cluster Manager to the broker.
• Line 3: Attach one Connection Manager to the Cluster Manager.
• Line 6: Create a list of cluster nodes. Note: you can use the same list for all nodes.
• Line 7: Install the list.
• The magic starts.

In this example, we implicitly create a Multiplex Transport Layer (MTL) instance when creating the Cluster Manager. A more advanced setup would instead create a cluster as follows:

mtl=require"smq.mtl".create("cluster password")
conn,err=require"smq.conn".create(mtl, 1900)
if conn then
   local list={"192.168.1.100","192.168.1.101", "192.168.1.102"}
   conn:setlist(list) -- Connect the nodes
end

smq_1=require"smq.hub".create()
cluster_1=require"smq.cluster".create(smq_1, mtl)

Figure 2: Creating MTL, Conn, and one SMQ/Cluster Combo

The above setup is required if you plan on creating multiple SMQ broker instances and having each broker instance be part of the cluster solution.

Download Cluster Example Code from GitHub.

SMQ Cluster Manager

The Cluster Manager adds clustering support for the SMQ broker, thus enabling any number of SMQ brokers to scale up horizontally. Note that the Cluster Manager requires a Multiplex Transport Layer (MTL) instance. You can either create an MTL instance or let the Cluster Manager create one for you. You must explicitly create an MTL if you plan on creating multiple SMQ brokers and SMQ Cluster Managers. Each broker/cluster-manager combo uses the same MTL instance.

SMQ Cluster Manager API

create(smq [,mtlOrPassword [,op]])

Create an SMQ Cluster Manager: require"smq.cluster".create(smq, mtlOrPassword, op).

Parameters

• table smq - The SMQ broker. A broker can have only one Cluster Manager.
• table|string|nil mtlOrPassword - An existing MTL instance, or a password for creating one. Omit it to create an MTL with an empty password.
• table|nil op - Optional settings. When creating an MTL implicitly, also accepts the MTL options.
  • string|nil name - The channel registration name, which must match on the corresponding remote brokers and be unique within a shared MTL. Defaults to SMQ-1, SMQ-2, and so on in creation order. Set an explicit name when creation order can differ between nodes.

Return values

• table cluster - The Cluster Manager. Construction does not establish peer connections; use a Connection Manager or commence sockets on its MTL.

Throws

Throws if the broker already has a Cluster Manager, the MTL channel name is already registered, MTL settings are invalid, or arguments are otherwise used incorrectly.

cluster:publish(data, [ptid,] subtopic)

Publish a message to the server SMQ client (server's ephemeral TID) on every connected cluster node.

When a subtopic is first used on a peer connection, its announcement and publication are sent together so another publication cannot separate them. This also applies to pubon() and publications forwarded by the broker.

Parameters

• string|table data - Message bytes or a Lua table to encode as JSON.
• number ptid - optionally publish on behalf of 'ptid'. The publisher tid is set to the ephemeral TID of the server SMQ client if the ptid is not set.
• string|number subtopic - sub-topic name or a sub-TID.

Use publish(data, subtopic) for the server's own publisher ID, or publish(data, ptid, subtopic) for an explicit publisher ID. In the latter form, pass 0 as subtopic when none is needed.

Return values

• number|boolean count - The number of cluster connections attempted, or false if an explicit publisher ID is unknown or belongs to a phantom peer owned by this cluster manager. A count of 0 means there are no connected cluster nodes. The count includes failed writes and does not confirm delivery.

This method cannot provide publication success information in its return values. Transport failures are not reported to the caller, and a positive count must not be treated as a success acknowledgement.

Throws

Incorrect message, publisher ID or subtopic arguments can throw during lookup, JSON encoding or frame assembly. Transport error returns are not forwarded to the caller.

cluster:pubon(data, ptid, topic, subtopic)

Publish on behalf of 'ptid' - publish on behalf of another SMQ client connected to the SMQ broker instance registered with the Cluster Manager.

Parameters

• string|table data - Message bytes or a Lua table to encode as JSON. The encoded payload must not exceed 65,520 bytes, so it can be forwarded to ordinary SMQ clients.
• number ptid - set sender of message.
• string|number topic - a topic name or a TID.
• string|number subtopic - sub-topic name or a sub-TID.

Return values

• boolean accepted - false if the publisher ID is unknown or belongs to a phantom peer owned by this cluster manager; otherwise true. A true result does not confirm delivery or that any remote node subscribes to the destination. Transport error returns are not forwarded to the caller.

This method cannot provide publication success information in its return values. The boolean reports publisher eligibility, not successful transmission or delivery.

Throws

Invalid message, topic or subtopic arguments can throw. For an accepted publisher, JSON encoding failure or an encoded payload larger than 65,520 bytes throws before transmission. This size limit does not apply to server-to-server cluster:publish().

cluster:close([msg])

Close this cluster manager's named MTL registration and notify connected peers using the same registration name.

Warning: This operation causes the notified brokers to disconnect their SMQ clients, but leaves clients connected to the initiating broker. This behavior is intentional. To disconnect the initiating broker's clients as well, explicitly call its smq:shutdown() method. Closing a cluster registration does not shut down the shared MTL transport.

Limitation: close() retains the cluster's connection state and the hub's publication callbacks. Subsequent local publications can therefore still attempt forwarding over the retained cluster connections. Peers that have closed the corresponding MTL registration discard these messages. Do not use close() as a guarantee that outgoing cluster publication traffic has stopped.

Parameters

• string|nil msg - Optional reason, defaulting to "close". Notified brokers send a nonempty reason to their clients in DISCONNECT messages, which tell well-behaved clients not to reconnect automatically. An empty string closes those connections without DISCONNECT, allowing their normal reconnect policy.

Return values

No values. This method does not confirm that each remote broker received the notification.

Throws

An invalid reason can throw when the close message is assembled. Transport error returns are not forwarded to the caller.

SMQ Multiplex Transport Layer

The Multiplex Transport Layer (MTL) provides communication busses for communication with other servers. The MTL is required when using the SMQ Cluster Manager, but the MTL can also be used for custom communication between connected servers.

The Multiplex Transport Layer is typically used in combination with the Connection Manager; however, the Multiplex Transport Layer can also be used directly if a more exotic clustering configuration is required than what can be provided by the Connection Manager. An example of such a connection configuration can be found in the example below. You can also combine the use of the Connection Manager with your own connection manager.

The following example shows how to establish an MTL connection by initially using HTTP. You can also establish an HTTPS connection by using an https:// URL, or use the proxy/tunneling settings available to the HTTP client library.

Common code required by both client and server code below (in a .preload/.config script).

-- Use the same private password on both nodes.
mtl=require"smq.mtl".create("my password")
function mtlstatus(peerAddr, sock, up, err)
   trace("MTL", peerAddr, up, err)
end

-- Replace the URL with the other node's address and LSP path.
ba.thread.run(function()
   local http=require"httpc".create()
   local ok,err=http:request{
      url="http://other-node.example/path/2/cluster/page.lsp",
      method="GET",
      header={SMQ="CLUSTER"}
   }
   if not ok then trace("HTTP request failed",err); http:close(); return end
   local status,err=http:status()
   if status ~= 204 then
      trace("MTL response rejected",status,err)
      http:close()
      return
   end
   local sock,data=ba.socket.http2sock(http)
   if not sock then trace("Socket transfer failed",data); http:close(); return end
   -- data may contain greeting bytes already received with the HTTP response.
   if not mtl:commence(sock,mtlstatus,data) then trace("MTL connection declined") end
end)

<?lsp
if request:header"SMQ" == "CLUSTER" then
   response:setstatus(204)
   if not response:flush() then return end
   local sock,data=ba.socket.req2sock(request,true)
   if not sock then trace("Socket transfer failed",data); return end
   -- Transfer any bytes already received after the HTTP request headers.
   if not app.mtl:commence(sock,app.mtlstatus,data) then trace("MTL connection declined") end
   return
end
response:senderror(404)
?>

References: Module httpc, ba.socket.http2sock, ba.socket.req2sock

Multiplex Transport Layer (MTL) API

create(password,[op])

Creates and returns an MTL instance: require"smq.mtl".create(password, op)

The API provided by the MTL is used by the Connection Manager and the Cluster Manager. The API may also be used by custom applications that wish to establish communication channels with the cluster nodes.

You do not need to understand the API returned by create() unless you use the MTL for your own custom transport.

Parameters

• string|nil password - Optional; nil selects an empty password. The password must be equal on all connected cluster nodes. The password can be any length and includes any characters (binary data). The password is sent encoded using HMAC-SHA512 and seeded by both peers, thus the password is protected on non-secure (non-TLS) connections.
• table|nil op - Optional configuration options:
  • function log(msg, important) - communication status is, by default, sent to the trace buffer. You may override this behavior by setting a log callback function. Argument msg is a string and is the message sent by the MTL instance or the associated Connection Manager. Argument 'important' is true if the message should be observed immediately by a person. The important flag is, for example, set if a cluster connection goes down. This would typically indicate that another server instance went down.
  • function onstatus - called when a connection is established and if the connection goes down. function onstatus(ipaddr, status, err): string ipaddr identifies the peer, boolean status is true when the connection is up and false when it goes down, and string|nil err gives the disconnect reason. Return values are ignored.
  • number keepalive - the keepalive and keepintv are socket options. The default keepalive time is 20.
  • number keepintv - the default keepintv socket option is 2.
  • table ping - send hard pings instead of using the socket option keepalive. The value for the attribute ping must be a table with the two following values:
    • number timespan - the time between sending ping messages. Minimum value is 4 * timeout.
    • number timeout - the maximum time in seconds to wait for the ping response and before the connection is considered dead/inactive. The minimum value is 2. The caller must ensure that timeout * 1000 is an integer no greater than 4294967295; an invalid interval throws when the ping timer is configured. The connection is terminated if a ping response is not received within this time limit.
    • function|nil roundtripCB - Optional callback roundtripCB(sock, milliseconds). The socket sock identifies the peer. The number milliseconds is the measured reply time, or nil when the peer times out. Return values are ignored. PONG replies are ignored unless a response is pending for that peer, including replies received when hard pings are disabled.

Return values

• table mtl - The new transport instance.

Throws

Throws for invalid ping settings, a non-function log option, or other incorrect argument use. User callbacks must be valid functions; their errors are not converted to transport error returns.

mtl:open(name,statusCB,dataCB)

Open a communication channel. You may register a channel while a peer handshake is pending. MTL announces it when that connection becomes ready; connected peers are notified immediately.

Parameters

• string name - each channel must use a unique name.
• function statusCB(sock,up[,idOrReason]) - connect/disconnect status information callback function:
  • socket sock - the peer's socket object.
  • boolean|nil up - boolean value set to true when the connection is established and false when the connection goes down. The value is set to nil should a peer call method mtl:close.
  • number|string|nil idOrReason - When up is nil, this is the peer's optional close reason (string), or nil if none was supplied. When up is false, this argument is absent. When up is true, this is the numeric multiplex handle to use when sending data frames with sendframe.
• function dataCB(sock,data) - a cluster node sent this node data.
  • socket sock - the sender's socket object.
  • string data - the data sent by the peer with a 6 byte header, thus the payload data starts at index position 7. You may extract the payload data as follows: data=data:sub(7)

Return values

• boolean ok - True after registering the channel. This does not confirm a connection or delivery of its announcement.

Throws

Throws if name is already registered or arguments are used incorrectly. Errors raised by a status callback propagate.

mtl:isopen(name)

Check whether this MTL instance has a local channel registration.

Parameters

• string name - The channel name.

Return values

• boolean open - True if registered, otherwise false. This does not indicate whether a peer is connected.

Throws

Does not throw for an unknown name.

mtl:close(name [,msg])

Remove a local channel registration and send a close notification to connected peers. A channel closed while a peer handshake is pending will not be announced when that peer connects. This leaves the shared transport open.

Parameters

• string name - The name used when calling mtl:open.
• string|nil msg - Optional reason passed to the peer's channel status callback with up=nil. An omitted or empty reason becomes nil in that callback.

Return values

• boolean result - Always false, including when a registration was removed. The return value does not report delivery of the close notification.

Throws

Incorrect argument use can throw. An unknown channel name is ignored.

mtl:commence(sock, statusCB [,data])

Start the initial handshake with the peer and commence normal operation if the handshaking succeeds. The up=true status callback runs after MTL has attempted its initial channel announcements, so opening a channel from that callback does not repeat its announcement in the initial loop. The Connection Manager calls mtl:commence for each connection it establishes with another server.

Parameters

• socket instance sock - a connected socket object, either from a client connection or an accepted socket from a server connection.
• function statusCB(peerAddr, sock, up [,err]) - connect/disconnect status information callback function:
  • string peerAddr - The peer's IP address.
  • socket sock - the peer's socket object - the object passed into mtl:commence.
  • boolean up - boolean value set to true when the connection is ready and false when the connection goes down. The value may be set to false without first receiving a true value if the handshaking fails. For example, it fails if the peer's password is not the same as this node's password.
  • string|nil err - Error information when up is false. This includes socket errors and "protocolerror" for an MTL frame length smaller than its six-byte header, an OPEN/CLOSE message missing its two-byte field, or a CLOSE message with an incomplete channel name. A cluster message missing its required type or numeric fields also reports "protocolerror". The same applies to conflicting subscription mappings, reuse of a remote subscription ID for a different topic, an unsubscribe for an unknown remote ID, or an acknowledgement for an unknown local subtopic. These errors close the peer connection and notify every named channel sharing that connection with up=false. Cluster disconnect cleanup removes routes and remote publisher records associated with that peer. An identical repeated subscription (the same name and remote ID) is accepted without adding another route. TCP reads may split a header or payload; the reader assembles these pieces before delivering the frame. If the initial 100-millisecond greeting probe times out after receiving part of the greeting, MTL preserves those bytes for its existing two-second retry. Only a timeout is retried; other initial read or framing errors close the connection through the normal handshake-failure path.
• string|nil data - Optional bytes already read from the socket, such as the second result from ba.socket.http2sock() or ba.socket.req2sock(). Default: nil. Pass these bytes unchanged and in order; they may contain part of the greeting or multiple frames. MTL consumes them before reading more bytes from the socket.

Return values

• boolean accepted - True when socket processing is started, before the handshake completes. False if this MTL has been shut down or the peer address matches the local address; the socket is closed in either case.

Throws

Incorrect socket, callback or buffered-data use can throw. Handshake and transport failures are reported asynchronously to statusCB, rather than returned by commence. Errors raised by application callbacks are not converted to transport errors.

mtl:hascon(ipaddr)

Parameters

• string ipaddr - The peer IP address, using the form reported by the status callback.

Return values

• boolean present - True if a peer entry exists. This includes a connection whose handshake is still in progress; use the status callback to determine readiness. False if no entry exists.

Throws

Does not throw for an unknown address.

See also conn:status.

mtl:shutdown()

Stop the ping timer, close all peer sockets, including those still performing their handshake, and clear the peer entries. A pending handshake cannot activate a connection after shutdown. Subsequent commence calls reject and close their sockets.

Parameters

None.

Return values

None.

Shutdown does not report canceled handshakes through the commence status callback. After shutdown, hascon() returns false for every peer. Local named-channel registrations remain in the instance.

Throws

No argument-validation errors for a valid MTL instance. Callback errors during socket cleanup can propagate.

sendframe(sock, id, data)

Send a data frame: require"smq.mtl".sendframe(sock, id, data).

Parameters

• socket sock - The connected peer socket.
• integer id - The multiplex handle received by the channel status callback when up=true.
• string data - The payload, which may be empty. The four-byte frame length includes the six-byte header. Available memory and socket limits also apply.

Return values

• boolean|nil success - The result from socket:write: true for accepted or queued output, or nil on failure. This does not confirm receipt by the peer.
• string|integer|nil detail - A socket error string, or an asynchronous queue byte count, as returned by socket:write. May be absent on success.

Throws

Incorrect payload, ID, or socket use can throw during frame assembly or socket:write. Ordinary write failures are returned unchanged.

SMQ Connection Manager

The Connection Manager (CONN) automates the connection of cluster nodes. The CONN takes a list of names or IP addresses and attempts to connect to the remote clusters. The CONN also acts as a server and waits for other CONNs to connect. When a connection is established -- i.e. when a client or server socket object is created, function mtl:commence is called and the socket object is passed into the MTL. The Connection Manager automatically attempts to reconnect broken connections.

One can manually add cluster names or ip-addresses to the CONN one at a time or one can set a pre-defined list. The CONN is designed to detect addresses pointing to 'self'. This construction makes it possible to use the same list on all cluster nodes.

create(mtl, port [,op])

Creates and returns an SMQ Connection Manager instance: require"smq.conn".create(mtl, 1999)

One can create multiple CONNs for the same MTL instance. One CONN may, for example, establish non-secure communication links and another may establish secure (TLS) communication links.

The connections are, by default, non-secure. To enable secure communication (TLS), set op.shark to a SharkSSL client object and op.sshark to a SharkSSL server object. You must also set up a certificate for the SharkSSL server object and a certificate store for the SharkSSL client object. The signer (CA cert) of the server certificate must be added to the client's certificate store. Non-trusted connections are closed by the client.

Parameters

• table mtl - An MTL instance, or a Cluster Manager containing one.
• number port - the port number used by the TCP server connection. This is the port number used by the other cluster nodes when attempting to connect to this cluster node.
• table op - optional socket and Connection Manager options. The socket options are documented in ba.socket.bind and ba.socket.connect. Connection Manager options:
  • SharkSSL object sshark - required for secure connections. The SharkSSL object must be initialized as a TLS server.
  • function onstatus - called when a connection is established and if the connection goes down. function onstatus(ipaddr, status, err): string ipaddr identifies the peer, boolean status is true when the MTL handshake succeeds and false when the connection goes down or its handshake fails, and string|nil err gives the failure reason. Return values are ignored. Initial TCP connection failures are logged and retried, without invoking this callback.

Return values

• table|nil conn - The new Connection Manager, or nil if the listening socket could not be created.
• string error - Returned with nil; the error from ba.socket.bind.

Throws

Throws for an invalid MTL argument, missing sshark when shark is provided, or incorrect socket arguments. Application callback errors are not converted to connection error returns.

conn:add(addr [,port])

Add a cluster node to the outbound connection list. Names resolving to the same IP address identify the same entry.

Parameters

• string addr - Hostname or IP address of the remote node.
• integer|nil port - Remote Connection Manager port. Defaults to the port supplied to create.

Return values

• boolean added - True when added. False if resolution fails or the IP address is already listed. Adding a duplicate does not replace its port.

Throws

Incorrect argument use can throw. The port is stored without validation here; incorrect socket options or port values can fail when a connection is attempted. Resolution failure returns false.

conn:setlist(addrlist)

Replace the outbound connection list. An empty table removes the list. Existing MTL connections remain open.

Parameters

• table addrlist - Entries may use string hostname/IP keys with numeric port values, or numeric keys with string hostname/IP values. The latter use the port supplied to create. Both forms may be mixed.

Return values

• boolean accepted - True if every entry was accepted and resolved. False if an entry has an unsupported key/value type or a name cannot be resolved. On false, the previous list is retained.

Throws

A non-table addrlist or incorrect argument use can throw. Numeric port values are not range-checked here; invalid ports can fail when a connection is attempted. Unsupported entry types and resolution failures return false.

conn:status()

Parameters

None.

Return values

• table connections - A new table mapping string IP addresses to boolean connection states. True means an established MTL connection; false means a listed outbound node without an established connection. Includes established connections that are not in the outbound list, such as incoming connections. Editing this snapshot does not affect the manager.

Throws

No argument-validation or operational errors for a valid Connection Manager instance.

conn:shutdown()

Cancel connection retries and close the listening socket. Queued outbound attempts do not start after shutdown. An outbound attempt already underway may finish, but its returned socket is then closed instead of being passed to MTL. Established connections in the shared MTL remain open.

Parameters

None.

Return values

• boolean closed - The listening socket's close result: true when closure is requested, or false if already closed. This does not describe the shared MTL connections.

Throws

No argument-validation errors for a valid Connection Manager instance. Errors during socket cleanup can propagate.

Auto Discovery of Cluster Nodes

We can create a simple mechanism for auto discovering and connecting cluster nodes when the nodes are on the same network. We can easily create an auto discovery service that finds other nodes by sending UDP broadcast messages.

The following example discovers nodes on a network that permits UDP broadcasts. All nodes must use the same UDP discovery port; each advertises its own TCP Connection Manager port.

local function broadcastCosock(s,conn,udpport,tcpport)
   local s,err = ba.socket.udpcon{port=udpport}
   if s then
      s:setoption("broadcast", true)
      local msg=ba.socket.h2n(2, tcpport)
      local data
      while true do
         if not data then s:sendto(msg,"255.255.255.255",udpport) end
         data,err=s:read(5000,true)
         if data then
            if #data == 2 then
               -- 'err' is now sender's IP i.e. s:read is using recvfrom
               conn:add(err, ba.socket.n2h(2, data)) -- Add IP address
            end
         elseif err ~= "timeout" then
            break
         end
      end
   end
   trace("Unexpected error in broadcastCosock", err)
end

local function autodiscover(conn,udpport,tcpport)
   ba.socket.event(broadcastCosock,conn,udpport,tcpport)
end

function autodiscover(conn,udpport,tcpport)

Start a UDP cosocket that broadcasts the local TCP port initially and after a five-second receive timeout. Incoming datagrams restart the wait, so continuous traffic can postpone the next broadcast. Two-byte datagrams supply the advertised TCP port; other lengths are ignored. The sender IP and advertised port are passed to conn:add, whose return value is ignored.

Parameters

• table conn - Required; an SMQ Connection Manager instance
• integer udpport - Required; the UDP port to listen on and to send UDP broadcast messages on
• integer tcpport - Required; the port number used by the local SMQ Connection Manager (CONN) instance

Return values

No values. This example does not return a handle for stopping discovery.

Throws

Incorrect arguments can raise errors in the discovery coroutine when socket operations or conn:add execute. Socket creation failures and read errors other than timeout end discovery and are written to the trace output. Broadcast-send error returns are ignored. These asynchronous failures are not returned by autodiscover().

Using SMQ for Server to Server Communication

The SMQ broker includes an integrated client that can be used by server-side Lua code. The integrated client enables server-side Lua code to function as any other SMQ client. By setting up an SMQ cluster of at least two nodes, SMQ can be used for communication with other connected nodes. In fact, you may use SMQ for server-to-server communication only and not have any other clients connected.

When communicating with other connected cluster nodes, you may use standard named (one-to-many) publish/subscribe messages and send one-to-one messages by sending it directly to the ephemeral topic ID, the unique ID created for each client. See the One-to-one Communication introduction for more information.

Unlike a regular connected SMQ client, the client embedded in the SMQ broker has the hard-coded ephemeral topic ID (etid) one (1). The server-side SMQ client will receive messages from any client publishing to etid 'one' as long as the client is connected to the same broker. A client connected to another broker, including the server-side client, cannot directly publish to etid 'one' and expect this to be sent over the cluster connection to another broker node. For this to work, we must first discover the client in the node we want to communicate with and have the SMQ Cluster Manager set up a communication path.

Each SMQ client has a unique etid and the etid for the SMQ client on the server-side is one. When any SMQ client communicates over a cluster connection, the SMQ Cluster Manager sets up a phantom connection client on the other side of the cluster connection. The phantom connection gets its own unique etid, but this etid is different from the etid on the origin broker. The purpose of the phantom client is to trap messages sent to this client's etid, send them over the cluster connection, and then send them to the correct client. This logic is handled by the SMQ Cluster Manager.

The following example is fully functional and can be installed in a .preload script and run in Mako Server "as is", except that you must use IP addresses that work on your network. The example uses a one-to-many message called 'ping', and the server-side SMQ client subscribes to this message. When the client receives a message addressed to 'ping', it responds by sending the one-to-one message 'pong' to the sender. In the code below, the sender of the 'ping' message is one of the server-side clients connected to the cluster.

local mtl=require"smq.mtl".create("cluster password")
local conn,err=require"smq.conn".create(mtl, 1900)
assert(conn,err)
local list={"192.168.1.100","192.168.1.101", "192.168.1.102"}
conn:setlist(list) -- Connect the nodes
local smq=require"smq.hub".create()
cluster=require"smq.cluster".create(smq, mtl)

smq:create("ping", 2)
smq:createsub("ping", 3)
smq:createsub("pong", 4)

local function onping(data,ptid,tid,subtid)
   trace(string.format("%-20s %10X %d %d",data,ptid,tid,subtid))
   smq:publish("I am good, thanks!", ptid, "pong")
end

local function onpong(data,ptid,tid,subtid)
   trace(string.format("%-20s %10X %d %d",data,ptid,tid,subtid))
end

--smq:subscribe('self', {subtopic="ping",onmsg=onping})
smq:subscribe('ping', {onmsg=onping})
smq:subscribe('self', {subtopic="pong",onmsg=onpong})

local function oneshot()
   for ip,stat in pairs(conn:status()) do
      trace(ip, stat and "connected" or "broken")
   end
   --cluster:publish("How are you?", "ping")
   smq:publish("How are you?", "ping")
end

ba.timer(oneshot):set(3000,true)

Figure 3: Server to Server Communication Example

Line 1 to 7, which is copied from Figure 2, sets up an SMQ cluster.

Lines 9 to 11 are not required, but this hard coding of topic and subtopic names makes the example printouts easier to understand. You may remove these three lines and the program will still work, but the topic IDs (tids) will be assigned random number values.

Function 'onping' on line 13 is registered as the callback when we subscribe to the named topic 'ping' on line 23. This function prints the data and responds by publishing 'pong' to the sender of the message, i.e., by sending a message to the publisher's ephemeral topic ID (ptid). Notice that we publish to the named topic ping, but the 'pong' message is named using a subtopic name. Different types of one-to-one messages can only be differentiated by using subtopic names. See subtopic names introduction for more information.

Function oneshot on line 26 is a timer function that starts 3 seconds after startup, which is more than the time needed for connecting the three cluster nodes. The function prints out the connection status of the two other connected nodes and then publishes a message to the named topic 'ping'.

In the cluster configuration on line 4, we set expectations for three cluster nodes, which means that each cluster node expects to connect to two other nodes. The printouts from one of the cluster nodes are shown below. Notice that function trace also prints out the code line. As an example, the first two printouts below are from code line 28.

28: 192.168.1.100      connected
28: 192.168.116.1      connected
onmsg 14: How are you?                  1 2 0
onmsg 14: How are you?           3C230B05 2 0
onmsg 14: How are you?           5E9E1C6E 2 0
onmsg 19: I am good, thanks!            1 1 4
onmsg 19: I am good, thanks!     3C230B05 1 4
onmsg 19: I am good, thanks!     5E9E1C6E 1 4

From these printouts, you can see that line 14 is activated three times. The first printout is received from 'self' since ptid is one. Recall that the server-side SMQ client's etid is hard-coded to one. The next two printouts are ping messages from the other two nodes. The printed ptid values (3C230B05 and 5E9E1C6E) are from phantom connection clients handled by the Cluster Manager. The phantom connection client enables us to send a one-to-one message on line 15 in figure 3 and directly respond to the sender of the message, which is another server-side SMQ client in another node.

You may have noticed that we received three ping messages, including the one sent from the same cluster node. We can add a filter in the onping callback to detect messages sent from 'self' by adding an if ptid ~= 1 clause and use this code to filter out messages received from 'self'. However, there is a better way to prevent sending messages to 'self'. In the code in Figure 3, we enable inter-cluster communication by using message 'ping' as a broadcast (one-to-many) message to discover other server-side SMQ clients in connected nodes. Instead of publishing to a named topic (ping), we can use the specialized publish methods in the SMQ Cluster Manager API for setting up one-to-one communication channels.

Notice the two code lines (22 and 30) that are commented out. You can enable those two code lines and comment out line 23 and 31. Line 30 uses the specialized cluster:publish method to send a message to the server-side SMQ client in all other connected nodes, but excludes 'self'. The next printout below is from this code setup.

28: 192.168.1.100      connected
28: 192.168.116.1      connected
onmsg 14: How are you?           ED981D9D 1 3
onmsg 14: How are you?           680F3958 1 3
onmsg 19: I am good, thanks!     680F3958 1 4
onmsg 19: I am good, thanks!     ED981D9D 1 4

From these printouts, you can see that we are no longer receiving the 'ping' message sent to 'self'. Also, notice that the 'ping' message type is now sent as a subtopic name.

We recommend that you analyze these printouts and cross-check them with the example code in Figure 3 to learn more about named topics and ephemeral topic IDs.