The Simple Message Queue (SMQ) Lua client connects a BAS, Mako Server or Xedge application to an SMQ broker. It publishes messages, subscribes to topics and observes subscriber counts. Create the client, install its callbacks, and wait for onconnect before subscribing or publishing. Close the client when its owning application unloads.
Connection and broker results arrive asynchronously through callbacks. Callback return values are ignored except where documented below. Callback exceptions are not caught by this module and can interrupt its connection or receive coroutine. Explicitly close owned clients; a running client can remain referenced by its asynchronous work, so garbage collection is not a substitute for application cleanup.
Create an SMQ client instance and initiate the connection with the broker.
Parameters
Note: the Lua client is similar to the JavaScript client, but you cannot subscribe or publish before the "onconnect" callback is called. You must re-subscribe to topics should the connection go down.
-- Connect to cluster node2: https://simplemq.com/
local url="https://node2.simplemq.com/smq.lsp"
local op={trusted=true}
local smq=require"smq.client".create(url,op)
-- Run the timer based publish test
local timer
local function runTest(accepted)
if not accepted then trace("Subscription refused") return end
trace"Starting test"
timer=ba.timer(function()
for i = 1, 100 do
smq:publish({count=i}, "mytopic.json")
coroutine.yield(true) -- Wait for the next timer tick
end
trace("Exiting timer coroutine")
end)
timer:set(1, true, true)
end
-- Received message for topic 'mytopic.json'
local function onmsg(message)
-- Undecodable input is delivered here as a string.
if type(message)=="table" then trace(message.count)
else trace("Undecoded message",message) end
end
-- Wait for connection before subscribing
smq.onconnect=function(tid, rnd, ipaddr)
local op = {
datatype="json",
onmsg=onmsg, -- callback for topic 'mytopic.json'
onack=runTest, -- We start the publish test when subscribed
}
smq:subscribe("mytopic.json", op)
end
function onunload()
trace"Stopping SMQ test"
if timer then timer:cancel() end
smq:close()
end
trace"Starting SMQ test"
Return values
Throws
Throws for invalid constructor arguments. Connection, authentication and broker failures are reported to onclose, rather than as a synchronous constructor error result.
Pass a connector function instead of a URL to provide your own connection setup. The client calls it for each connection attempt from its connection coroutine. Complete each attempt exactly once by calling the supplied callback, either before the connector returns or later. Returning a socket or error from the connector does not complete the attempt.
Parameters
Return values
Ignored. Report the outcome through callback.
Throws
Errors raised by the connector can terminate the connection coroutine. Report ordinary connection failures through callback(nil, error), so the client can invoke onclose and apply its reconnect policy.
On success, transfer a connected socket ready to exchange raw SMQ frames. Complete any HTTP setup or other transport negotiation first. The client reads the broker's INIT message and performs SMQ authentication after this callback.
Parameters
Return values
None.
Throws
Incorrect socket or buffered-data use can throw during subsequent connection processing. The connector is responsible for calling callback exactly once; duplicate calls are not checked. Application callback errors follow the normal client callback behavior.
For deferred completion, callback can resume the connection coroutine before returning. If the client was closed while setup was pending, a subsequently supplied socket is closed without starting the SMQ handshake or invoking onconnect/onclose. Complete pending attempts even after client closure so their connection coroutine can finish.
The following lists the methods available on an SMQ Lua client instance. Most functions do not return a value immediately; instead, results are typically reported through optional callback functions. Only methods that explicitly document a return value return one directly.
Create a topic and resolve its numeric ID before publishing, for example when the broker applies topic authorization. Publishing to a topic name also requests its ID automatically. Create subtopics separately with createsub(); this method has no subtopic argument.
Parameters
onack(accepted, topic, tid). Its arguments are boolean accepted, string topic and number tid, the unsigned 32-bit topic ID. Use the ID only when accepted is true. Callback return values are ignored.If the name is already resolved, the callback runs synchronously before this method returns. Otherwise it runs after the broker acknowledgement.
Return values
Throws
Incorrect topic or callback usage can throw. Broker refusal is reported as accepted=false to onack.
Create a sub-topic and fetch the subtopic ID. The createsub method is typically used prior to publishing a message on a specific topic and sub-topic if the server logic implements authorization. Alternatively, the publish method may be used directly with topic and sub-topic strings, respectively. The publish method will manage the sequence and creation of a topic, then sub-topic, if you publish to a topic, sub-topic, or both that are unknown to the client stack.
Parameters
Return values
Throws
Incorrect subtopic or callback usage can throw. Broker refusal is reported as accepted=false to onsuback.
Gracefully close the connection. You cannot publish any messages after calling this method.
Parameters
None.
Return values
No values. Closes an existing socket and prevents continued use of the client. Repeated calls have no effect. Explicit close does not call onclose. The same method is used by __close and __gc.
Throws
An invalid receiver is incorrect API usage. The method does not deliberately throw for socket-close failures and does not return them.
Get the client's ephemeral topic ID. Each client is assigned a unique topic ID, and this topic ID is included when publishing a message. All subscribers receiving the message published by a client can use this (tid) for identification purposes or for sending messages directly to the "publisher" of the message. See the subscribe onmsg for example code.
Parameters
None.
Return values
Throws
An invalid receiver is incorrect API usage. An unavailable topic ID is returned as nil.
Publish messages to a topic and optionally to a sub-topic. Topics may be topic names (strings), topic IDs (numbers), or ephemeral topic IDs (numbers). Messages publishing unresolved topic names are temporarily queued. Pending topic names are resolved by calling create or createsub and result in the message being de-queued and published if create or createsub reports are accepted by the server. Once the server denies a create or createsub report request, all messages sent to (unresolved) topic names are de-queued and silently discarded. You may consider using create or createsub prior to calling "publish" if the server solution implements topic authorization. Note: the maximum encoded payload is 65,520 bytes (0xFFF0). The SMQ header uses another 15 bytes, giving the protocol's maximum message length of 65,535 bytes. Oversized messages throw during encoding instead of wrapping the length field.
Parameters
Return values
Throws
Invalid topic arguments, values that cannot be JSON encoded, or values that cannot be encoded in the message can throw. A disconnected client returns false before encoding. Payload encoding and size checks run in publish(), before any topic or subtopic name-resolution request. Invalid or oversized payloads therefore throw in the calling function. Table data is encoded immediately into a string; later changes to the original table do not alter a pending publication.
Subscribe to a topic and optionally to a sub-topic. You can subscribe multiple times to the same topic if you use sub-topics. Subscribing to a topic without providing a sub-topic introduces a "catch all" for sub-topics that do not correspond to any subscribed sub-topics.
The topic name "self" is interpreted as subscribing to the client's own Ephemeral Topic ID -- in other words, it means subscribing to the (tid) returned by method gettid. Subscribing to your own Topic ID makes it possible for other connected clients to send a message directly to this client.
Subscription requests are stored temporarily if a subtopic name must be resolved before subscribing. They are not queued before a connection is established. Re-subscribe after reconnection.
A known topic ID does not imply a subscription. This method sends a subscription request even after create() has resolved the name, and after unsubscribe(). Pending topic-creation and subscription acknowledgements are handled separately. The special topic "self" is registered locally without a broker subscription request.
Calling unsubscribe() cancels outstanding subscription requests for that topic, including requests waiting for a subtopic ID. Cancelled requests do not invoke their pending onack callbacks or install onmsg handlers. A later subscribe() is a new request and is not cancelled by an earlier acknowledgement.
Parameters
Return values
No values. Wait for function settings.onack to learn whether a broker subscription was accepted. Ordinary subscriptions made while disconnected are not queued; subscribe from onconnect/onreconnect.
Throws
Incorrect topic, subtopic, settings or callback usage can throw. Broker refusal is reported through onack.
Requests the server to unsubscribe the client from a topic. All registered onmsg callback functions, including all callbacks for sub-topics, will be removed from the client stack.
Parameters
Return values
No values. Removes the local subscription and all its topic/subtopic handlers, and cancels pending subscription callbacks. This also applies when messages use only the global onmsg handler. If a broker subscription acknowledgement is pending, the client waits for it before sending the unsubscribe packet; a refused subscription needs no cancellation packet. If only subtopic resolution is pending, the cancelled request does not proceed to subscribe. A later subscribe() remains valid. Messages already in transit cannot be recalled and may reach the global onmsg handler when no topic handler matches.
Throws
Throws if a string topic name is neither locally known nor associated with a pending subscription, or if an ID that must be encoded is invalid. An unresolved pending subscription can be cancelled by name.
Requests the broker to provide change notification events when the number of subscribers to a specific topic changes. See the JavaScript client -> observe for details on using this method.
See the broker's SMQ:observe() documentation for details on using observe()
Parameters
Return values
No values. Registers onchange and sends the observation request. Observing this client's own ephemeral ID or an already observed ID has no effect. The existing callback is retained.
Throws
Throws if a string topic name is not locally known, or if an ID that must be encoded is invalid. An invalid callback can fail when notified.
Stop receiving change notifications for a topic or ephemeral topic ID.
Parameters
Return values
No values. Removes an existing observation and sends its cancellation; an ID that is not observed has no effect.
Throws
Throws if a string topic name is not locally known, or if an ID that must be encoded is invalid.
Translates topic ID to topic name.
Parameters
Return values
Throws
An invalid receiver is incorrect API usage. An unknown lookup key returns nil without throwing.
Translates topic name to topic ID.
Parameters
Return values
Throws
An invalid receiver is incorrect API usage. An unknown lookup key returns nil without throwing.
Translates sub-topic ID to sub-topic name.
Parameters
Return values
Throws
An invalid receiver is incorrect API usage. An unknown lookup key returns nil without throwing.
Translates sub-topic name to sub-topic ID.
Parameters
Return values
Throws
An invalid receiver is incorrect API usage. An unknown lookup key returns nil without throwing.
You can set the following event callback functions on an SMQ instance.
The onauth function must be set if the server requires authentication. The event function is called during the SMQ handshaking phase and just before the client SMQ stack sends the CONNECT event to the broker.
Parameters
Return values
Throws
This is an application callback. Exceptions raised by it propagate in the calling coroutine; the SMQ client does not protect the call. Handle application failures inside the callback.
The onconnect function is called after a successful connection sequence and if the server accepted the credentials. If the connection was unsuccessful or if the connection was not accepted by the broker, the onclose function is called instead.
Parameters
Return values
No return values are used. Register subscriptions here. An optional onreconnect callback takes its place after an established connection is lost and restored.
Throws
This is an application callback. Exceptions raised by it propagate in the calling coroutine; the SMQ client does not protect the call. Handle application failures inside the callback.
The onmsg function receives raw data when no matching topic/subtopic callback is installed. JSON decoding failures stay with the topic callback as raw strings; they are not redirected here.
Parameters
Return values
No return values are used. The callback receives the original message bytes as a string.
Throws
This is an application callback. Exceptions raised by it propagate in the calling coroutine; the SMQ client does not protect the call. Handle application failures inside the callback.
The onclose function is called if the connection cannot be established, the server denied access, the server gracefully closed the connection, or if the connection unexpectedly closed.
Invalid message types, short fixed fields and acknowledgements without a matching request report string message="protocolerror" and close the connection. An invalid initial SMQ greeting reports "nonsmq". These received-data failures do not deliberately throw. Complete messages following the connection acknowledgement in the same socket read are retained for normal dispatch.
The onclose function can request the SMQ client to attempt to reconnect. Function onreconnect is called if the re-connect attempt is successful. The onclose function is called again if the re-connection attempt is unsuccessful.
Parameters
Return values
For a failed connection attempt, a positive elapsed time shorter than retryDelay is subtracted from the delay. If the elapsed time is zero or at least retryDelay, the full delay is used. For example, returning 2000 after a one-second attempt waits one more second; returning 2000 after a three-second attempt waits two more seconds. This initial-attempt timing is intentional.
Throws
This is an application callback. Exceptions raised by it propagate in the calling coroutine; the SMQ client does not protect the call. Handle application failures inside the callback.
The onreconnect function is called if the connection closed, the onclose function requested the SMQ client to reconnect, and if the re-connect was successful. Function onconnect is called if you do not provide an onreconnect.
Parameters
Return values
No return values are used. Re-register subscriptions and observations; the old session mappings and callbacks have been cleared.
Throws
This is an application callback. Exceptions raised by it propagate in the calling coroutine; the SMQ client does not protect the call. Handle application failures inside the callback.