Skip to content

Messaging

Messaging is a publish-subscribe framework bundled with Kx Stream. It runs in a server-client model with a messaging server process acting as the discovery service for processes that consume or publish data.

Processes register their topics and the server matches them with corresponding publisher or subscriber processes. Subscription topics are specified in JSON with a table and optional column filters. Both publishers and subscribers can specify filters. Helper APIs for generating JSON are described in the Template API guide.

Two Messaging server instances are bundled with Stream and can be run from the DS_launch_MS_{A|B} workflows. Most of the Stream process templates contain a messagingServer instance parameter, so to enable a process for messaging, set this to the DS_MESSAGING_SERVER:DS config parameter, which contains the names of the two Stream instances. On startup, processes will then register with the Messaging servers and be available to publish and subscribe.

Example topics

Topic Description
dxQuote Subscribe to or publish all messages in the dxQuote table. No filtering is applied at the publisher.
{"dxQuote":{"sym":"EURUSD"}} Subscribe to dxQuote table where sym=`EURUSD
{"dxQuote":{"sym":"EURUSD","src":"FD"}} Subscribe to dxQuote table for multiple sym=`EURUSD, src=`FD

In addition to matching on topics, Messaging has the concept of a channel. This is a symbol tag that provides an extra layer to distinguish between the same topics. For example, if two tickerplants were running hot-hot, the subscriber would want to receive the data only once. This would be done by specifying two different channels. When a publisher and subscriber have the same channel and their topics overlap, the Messaging server will match them and initiate handshakes.

Subscription topics

When two processes have been matched and handshakes completed, subscription state is initiated on the publisher process. Whenever that process calls the publisher APIs, it will check the data being published against the current list of subscriptions. The APIs all take table name and data as parameters and will find subscriptions for the same table first. It will then apply any subscription filters to the data and publish the results.

There are multiple filtered topic modes that define how the publisher sends the data. The following sections describe them in detail. The APIs for each mode are detailed in the Template API Guide.

Segmented

In segmented mode, when multiple filter values are applied on the same column, the topic is split into separate subscriptions on the publisher. So three column values would result in three separate topics. If the publisher sent data with mixed column values, they would be filtered and published in separate batches. This is used when the subscriber wants to do bulk operations on one value at a time.

Multiple filters on one column: : JSON: {"dxQuote":{"sym":["EURUSD","GBPUSD"],"src":"FD"}} : dxQuote where sym=`EURUSD, src=`FD : dfxQuote where sym=`GBPUSD, src=`FD

Multiple filters on more than one column: : JSON: {"dxQuote":{"sym":["EURUSD","GBPUSD"],"src":["FD","Kx"]}} : dxQuote where sym=`EURUSD, src=`FD : dfxQuote where sym=`GBPUSD, src=`FD : dxQuote where sym=`EURUSD, src=`Kx : dfxQuote where sym=`GBPUSD, src=`Kx

Bulk

In bulk mode, when multiple filter values are applied to the same column, the topic is not split. In this mode, subscribers will receive data filtered as specified but with all messages in one batch per topic.

Multiple filters on one column: : JSON: {"dxQuote":{"sym":["EURUSD","GBPUSD"],"src":"FD"}} : dxQuote where sym in `EURUSD`GBPUSD, src=`FD

Multiple filters on more than one column: : JSON: {"dxQuote:{"sym":["EURUSD","GBPUSD"],"src":["FD","Kx"]}} : dxQuote where sym in `EURUSD`GBPUSD, src in `FD`Kx

Sharded

This mode is supported for subscriptions only and is used when subscribers want to shard (split) data by a single column across multiple processes. For instance for load balancing, one process might take responsibility for stocks beginning A-K and a second for L-Z. The important thing to note is that the comprehensive list of instruments might not be known ahead of time and could change. Maintaining the list and dynamically updating subscriptions could be tricky to implement.

Instead, Messaging allows the developer to specify a shard column and regex condition to implement the split.
code.kx.com Cookbook/Regular expressions

This can be combined with other column filters, which are treated as bulk.

Shard filter only: : JSON: {"dxQuote":{"sym":[".q.like","[a-hA-H]*"]}} : dxQuote where sym like "[a-hA-H]*"

Shard and multiple column filter: : JSON: {"dxQuote":{"sym":[".q.like","[a-hA-H]*"],"src":["FD","KX"]}} : dxQuote where sym like "[a-hA-H]*", src in `FD`KX

Filters are currently supported only for symbol columns.

Topic hierarchy

Below is tabulated the combinations of subscription mode and publisher topics, given all the combinations of publisher, subscriber and topic type from the previous section. The vertical axis is the publisher type and the horizontal is subscriber type.

Subscribers have two other modes that aren’t supported for publishers:

  • sharding is supported only for subscribers
  • subscribers can also send a blank topic (empty string) to subscribe for all tables with no filters
blank no filter segmented bulk shard
No filter Match with no filter Match with no filter Segmented match of all syms Bulk match of all syms Shard match of all syms
Segmented Segmented match of all syms Segmented match of all syms Segmented match of overlapping syms Segmented match of overlapping syms Segmented match of overlapping syms
Bulk Bulk match of all syms Bulk match of all syms Segmented match of overlapping syms Bulk match of overlapping syms Bulk match of overlapping syms

Combinations of publisher types (vertical) and subscriber types (horizontal)

Reconnections

When a connection to another component drops, by default the process will try to reconnect on a timer. This behavior is configurable via enhanced instance parameters. The table of parameters below details what options are available.

parameter type default description
.dm.i.retryEnabled boolean true Reconnect when messaging component drops
.dm.i.retryInterval int 10000 Interval between reconnect attempts (milliseconds)
.dm.i.retryTimeout int 500 hopen connection timeout (milliseconds)
.dm.i.retries int 10

Table callbacks

The ability to specify function callbacks against a table is available in subscribers. Each table can have one or more functions to be executed whenever an update is received.

The standard model would be to;

  • setup the table to function(s) callbacks on instance startup
  • apply the callbacks whenever a new message is received to the upd or realTimeMsgFunct

This allows multiple instances to share one generic upd function but specify different callbacks per process.

An example of the API calls is provided below.

 callback:{[t;x] 0N!(t;x) };
 trade:([] time:.z.p; sym:`A`B; price:100 200.; size:1000 1000)

 // add a callback
 .dm.addCallback[`trade; `callback];

 // execute callbacks
 .dm.applyCallbacks[`trade; trade];
 /=> (`trade;+`time`sym`price`size!(2018.08.06D05:42:44.188299000 2018.08.06D05:42:44.188299000;`A`B;100 200f;1000 1000))

 // remove callback
 .dm.removeCallback[`trade; `callback];

Performance tuning

The Messaging server template contains two parameters that can help improve messaging performance.

useUnixDomainSockets : Enables the Unix Domain Socket functionality of kdb+. If publishers and subscribers reside on the same server and this option is enabled, they will communicate via domain sockets. This can show considerable efficiency gains over standard TCP/IP.

broadcastPublish : Improves performance of publishers if multiple subscriptions exist for the same topic. Instead of the publisher serializing the same message for each duplicate subscription, it will serialize only once and broadcast-publish to multiple handles. Under the covers it maps to the -25! operator in kdb+.

Both modes are disabled by default.

Event callbacks

Callbacks can be set up for various messaging events. This can be accomplished via the INSTANCE_CONFIG parameter and overrides.

Enhanced Instance Configuration under the Developers section of the Kx Control document for details of how to set up an INSTANCE_CONFIG override for a process.

callback event parameter type
.dm.cb.connect When a connection is made to another process remoteId integer
dict (keys host, port, tls, UDS) dictionary
dict[`host] symbol
dict[`port] integer
dict[`tls] symbol
dict[`UDS] boolean
.dm.cb.addSub (RTE) When a subscription is added to a publisher subscriberId integer
dict (keys topic, channel, broadcast) dictionary
dict[`topic] mixed list
dict[`channel] symbol list
dict[`broadcast] boolean
.dm.cb.addSub (TP) When a subscription is added to a publisher subscriberId integer
dict (keys topic, channel, broadcast, logFile, numMsgs, schema, logStreamer, instance, logFileH, numMsgsH, intradayLogFiles, intradayTables) dictionary
dict[`topic] mixed list
dict[`channel] symbol list
dict[`broadcast] boolean
dict[`logFile] symbol
dict[`numMsgs] long
dict[`schema] mixed list
dict[`logStreamer] symbol
dict[`instance] symbol
dict[`logFileH] symbol
dict[`numMsgsH] long
dict[`intradayLogFiles] symbol list
dict[`intradayTables] symbol list
.dm.cb.connected When connection to another process is confirmed remoteId integer
.dm.cb.deleteSub When a subscription is removed from a publisher subscriberId integer
dict (keys topic, channel) dictionary
dict[`topic] string
dict[`channel] symbol
.dm.cb.disconnect When a connection is removed due to dropped connection remoteId integer
.dm.cb.receiveConn When a connection is received from another process remoteId integer
dict (keys host, port, tls, UDS) dictionary
dict[`host] integer
dict[`port] integer
dict[`tls] symbol
dict[`UDS] boolean
.dm.cb.receiveReconn When a reconnection is confirmed on a process being reconnected to remoteId integer
.dm.cb.reconnect When a reconnection is registered on a process making the reconnection remoteId integer
dict (keys host, port, tls, UDS) dictionary
dict[`host] integer
dict[`port] integer
dict[`tls] symbol
dict[`UDS] boolean
.dm.cb.subConfirm (TP) When a successful subscription is confirmed publisherId integer
dict (keys topic, channel, broadcast) dictionary
dict[`topic] mixed list
dict[`channel] symbol list
dict[`broadcast] boolean
.dm.cb.subConfirm (RTE) When a successful subscription is confirmed publisherId integer
dict (keys topic, channel, broadcast, logFile, numMsgs, schema, logStreamer, instance, logFileH, numMsgsH, intradayLogFiles, intradayTables) dictionary
dict[`topic] mixed list
dict[`channel] symbol list
dict[`broadcast] boolean
dict[`logFile] symbol
dict[`numMsgs] long
dict[`schema] mixed list
dict[`logStreamer] symbol
dict[`instance] symbol
dict[`logFileH] symbol
dict[`numMsgsH] long
dict[`intradayLogFiles] symbol list
dict[`intradayTables] symbol list
.dm.cb.subDeleted When confirmation that a subscription has been deleted is received publisherId integer
dict (keys topic, channel) dictionary
dict[`topic] string
dict[`channel] symbol