Messaging
This module contains functions for interfacing with the pub/sub framework of Control. Components register as sources of or subscribers to streams of data. Messaging-enabled processes will register their interested topics with a set of Messaging Servers. These are discovery processes that match publishers to subscribers for a given topic. Once the source and subscribers are matched, a set of APIs is provided is provided for the publisher to push data between them.
For more information, please see the KX Stream documentation.
Subscription Modes
APIs for each of the subscription modes.
Segmented
- .dm.regsub
- .dm.regsubc
- .dm.regsrc
- .dm.regsrcc
Bulk
- .dm.regsubbulk
- .dm.regsrcbulk
Sharded
- .dm.regsubshard
.dm.bdc.disable
Disables broadcast messaging for the current process. If broadcast publishing is enabled at the messaging server level, all processes will use it by default. This provides an override to disable it if required.
Example:
.dm.bdc.disable[]
.dm.setRegHost
Sets the host to use for messaging. Should be called prior to registration with the messaging server.
The API should be called as part of an instruction run on startup. In order to run the it before registration, the instruction
should be added to the .al.preStartupInstructions parameter of the INSTANCE_CONFIG override.
Parameter:
| Name | Type | Description |
|---|---|---|
| host | symbol | Registration host |
Example:
.dm.setRegHost[`host]
.dm.addCallback
Adds a callback function against a table. Function must have two parameters;
- table name (symbol)
- data.(table)
Parameters:
| Name | Type | Description |
|---|---|---|
| table | symbol | Table name |
| function | symbol | Function name |
Example:
.dm.addCallback[`dxCPUStats; `updCPU]
.dm.applyCallbacks
Applies any function callbacks for incoming topic and data
Parameters:
| Name | Type | Description |
|---|---|---|
| table | symbol | Table name |
| data | table | Message data |
Example:
updCPU:{[t;x] 0N!(t;x)}
data:([] time:.z.p; sym:`server1`server2; cpu:20 40f);
.dm.addCallback[`dxCPUStats; `updCPU]
.dm.applyCallbacks[`dxCPUStats; data]
.dm.removeCallback
Removes a callback function against a table.
Parameters:
| Name | Type | Description |
|---|---|---|
| table | symbol | Table name |
| function | symbol | Function name |
Example:
.dm.removeCallback[`dxCPUStats; `updCPU]
.dm.pubnoreply
Publishes a table name and data to subscribers except calling handle (handle=.z.w).
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pubnoreply[`trade; data]
.dm.pub
Publishes a table name and data to all subscribers. Will apply any column or filtering for each subscription.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pub[`trade; data]
.dm.pubc
Publishes a table name and data to all subscribers on a given channel. Will apply any column or filtering for each subscription.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pubc[`channel; `trade; data]
.dm.pubdata
Publishes a table name and data to subscribers. See .dm.pub for full details. Published messsage has the format
(`.u.upd; `table; (col1...; col2..; col3...))
Publishing data in the list format results in a smaller message and performance gains can be seen in some circumstances. However filtering will only be applied at the table level.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | list | List of lists of column data |
Example:
data:(2#.z.p; `$("EUR/USD";"EUR/GBP"); `DB`UBS);
.dm.pubdata[`trade; data]
.dm.pubdatac
Publishes a table name and data to subscribers on a given channel. See .dm.pub for full details. Published messsage has the format
(`.u.upd; `table; (col1...; col2..; col3...))
Publishing data in the list format results in a smaller message and performance gains can be seen in some circumstances. However filtering will only be applied at the table level.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | list | List of lists of column data |
Example:
data:(2#.z.p; `$("EUR/USD";"EUR/GBP"); `DB`UBS);
.dm.pubdata[`channel; `trade; data]
.dm.pubflush
Publishes a table name and data to subscribers. Will flush each handle after publishing to ensure the data is sent immediately.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pubflush[`trade; data]
.dm.pubflushc
Publishes a table name and data to subscribers on a given channel. Will flush each handle after publishing to ensure the data is sent immediately.
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pubflushc[`channel; `trade; data]
.dm.pubmult
Publishes multiple tables downstream in a single message. Messages have the format
q
(`updM; `table1`table2`table3; ( data1 ; data2 ; data3 ))
Filtering is only applied at the table level
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol[] | Table names |
| x | table[] | List of table data |
Example:
t1:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
t2:flip `time`sym`cond!(.z.p; `$("EUR/USD";"EUR/GBP"); "AD");
.dm.pubmult[`trade`quote; (t1; t2)]
.dm.pubmultc
Publishes multiple tables downstream in a single message to a given channel. Messages have the format;
q
(`updM; `table1`table2`table3; ( data1 ; data2 ; data3 ))
Filtering is only applied at the table level
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol[] | Table names |
| x | table[] | List of table data |
Example:
t1:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
t2:flip `time`sym`cond!(.z.p; `$("EUR/USD";"EUR/GBP"); "AD");
.dm.pubmultc[`channel; `trade`quote; (t1; t2)]
.dm.pubnoreplyc
Publishes a table name and data to subscribers on a given channel except calling handle (handle=.z.w).
Parameters:
| Name | Type | Description |
|---|---|---|
| t | symbol | Table name |
| x | table | Data to be published |
See Also: .dm.pubnoreply
Example:
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pubnoreplyc[`channel; `trade; data]
.dm.buildSubscriptionString
Creates a JSON messaging topic from a table and filter dictionary
Parameters:
| Name | Type | Description |
|---|---|---|
| table | symbol | Table name |
| filter | dict | Dictionary of column names and filters |
Example:
.dm.buildSubscriptionString[`trade; `sym`src!(`$"EUR/USD"; `UBS`DB)]
/=> "{\"trade\":{\"sym\":\"EUR/USD\",\"src\":[\"UBS\",\"DB\"]}}"
.dm.regsrc
Register as a publisher of a topic with a blank channel
Parameter:
| Name | Type | Description |
|---|---|---|
| t | string | Topic |
Example: Publisher of trade
.dm.regsrc["trade"]
.dm.regsrcbulk
Registers as a bulk publisher of a channel and topic
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | string | Topic |
Example: Register bulk sub for trade table
syms:`$("EUR/USD";"USD/JPY";"GBP/USD");
topic:.dm.buildSubscriptionString[`trade; enlist[`sym]!enlist[syms]]
.dm.regsrcbulk[`tickerplant; topic]
.dm.regsrcc
Register as a topic source on a specific channel
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | string | Topic |
Example: Publisher of trade on tickerplant channel
.dm.regsrcc[`tickerplant; "trade"]
.dm.regsub
Register a subscription to a topic with a blank channel
Parameter:
| Name | Type | Description |
|---|---|---|
| t | string | Topic |
Example: Subscriber to trade
.dm.regsub["trade"]
.dm.regsubbulk
Registers bulk subscription of a channel and topic
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | string | Topic |
Example: Register bulk sub for trade table and three symbols
syms:`$("EUR/USD";"USD/JPY";"GBP/USD");
topic:.dm.buildSubscriptionString[`trade; enlist[`sym]!enlist[syms]]
.dm.regsubbulk[`tickerplant; topic]
.dm.regsubc
Register as a subscriber to a topic on a specific channel
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | string | Topic |
Example: Subscriber to trade on tickerplant channel
.dm.regsubc[`tickerplant; "trade"]
.dm.regsubshard
Registers as a sharded subscriber for a channel and topic. The topic acts like a bulk subscriber but with an additional sharding filter.
Hence the function signature is slightly different. It takes the usual elements of channel, table and filters but with an additional two params: column and filter.
These indicate what column to perform the sharding on and what the condition is.
An example shard would be sym like "[a-g]*". This would subscribe to a table for records where the sym
column begins with characters a-g.
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | symbol | Table |
| col | symbol | Shard column |
| shard | string | Shard topic |
| filt | dict | Dictionary of additional filters |
Example: Shard on sym
.dm.regsubshard[`tickerplant; `trade; `sym; "[a-g]*"; ()!()]
Example: Shard with additional filter
.dm.regsubshard[`tickerplant; `trade; `sym; "[a-g]*"; enlist[`src]!enlist `UBS`DB]
.dm.unsub
Unsubscribe from topic
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Unused parameter |
| t | string | Topic |
Example: Unsubscribe from trade topic
.dm.unsub[`; "trade"]
.dm.unsubbulk
Unsubscribe from a bulk channel and topic
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel |
| t | symbol | Topic |
Example:
syms:`$("EUR/USD";"USD/JPY";"GBP/USD");
topic:.dm.buildSubscriptionString[`trade; enlist[`sym]!enlist[syms]]
.dm.unsubbulk[`tickerplant; topic]
.dm.unsubc
Unsubscribe from topic and channel
Parameters:
| Name | Type | Description |
|---|---|---|
| c | symbol | Channel name |
| t | string | Topic |
Example: Unsubscribe from topic and channel
.dm.unsubc[`tickerplant; "trade"]
Callbacks
This section details the APIs for adding functional callbacks on subscribed topics. It includes APIs to add and trigger callbacks for specified tables.
Publishing
The APIs in this module related to publishing data. They are called on a publisher process and send data downstream to any subscribed processes.
The publisher will have a list of topics and associated process handles. The topics consist of table names and column filters.
Each API (except .dm.pubdata) takes a name and table of data. The table will be filtered for each topic so each downstream process
only receives the data it's interested in. The downstream consumer will receive a message in the form
(`upd; `table; (data))
.dm.pubdata is slightly different. Instead of the data being a table, it's a list of lists. It doesn't
perform any filtering on the received data and publishes to .u.upd instead of upd. This is used for publishing to tickerplants where throughput is the priority.
(col1[]; col2[]; ...; colN[])
Multi-publish
The ability to publish multiple tables in a single message is available via .dm.pubmult. This means the publisher only has to
serialize and send a single message instead of one per table. The message format is below.
(`updM; `table1`table2; ( table1 ; table2 ))
Row level filtering is not supported for multi-publish
API sets
There are five sets of publish APIs. In each set, the main API will publish to all channels and the other (with the trailing c suffix) will publish to a specific channel.
data:flip `time`sym`src!(.z.p; `$("EUR/USD";"EUR/GBP"); `DB);
.dm.pub[`trade; data]
.dm.pubc[`channel; `trade; data]
Null channel publish
When using the per-channel publish APIs, the behavior when using a null channel can be unexpected. See here for more information.
Registration
This section details the APIs for managing messaging topics. It includes APIs to build topics and register as a publisher or subscriber with the messaging servers.