Message Channels

A message channel is a named subscription-based broadcast bus. A strategy publishes to a channel by name, and every subscriber on that channel receives the message. Channels carry data between strategies, between strategies and external applications, and (optionally) between strategies and Kafka topics.

When to use them

Feeding external signals to a strategy (inflation prints, employment, VIX, Fed decisions).
Reconfiguring a running strategy.
Sharing computed signals between strategies (compute once, consume many times).
Coordinating behaviour across strategies.
Publishing data to external applications.
Publishing to or reading from Kafka.

Internal vs external channels

Any data type is allowed.
No serialisation: Java object references are published directly.
Minimal overhead.
Important: use immutable data types or otherwise ensure thread safety. Strategies on different threads can read the same object simultaneously.

Data type must be String.
Strategies serialise on send and deserialise on receive.
Reachable from outside via REST or WebSocket (see Strategy Server API).

Basic usage

A strategy receives messages on a channel by opening it. Opening a channel subscribes the strategy to messages on it.

interface StrategyContext {

  <T> MessageChannel<T> openChannel(Class<T> type, String name);

  // ...
}

Note

Every strategy that opens a given channel must agree on its data type. The first strategy to open the channel fixes the type. Opening with a different type fails with an exception.

When a new message arrives, the strategy receives an onMessage callback that names the channel. Pull the message with getMessage() on the channel.

Each Message carries:

A timestamp of when the message was published.
The origin (for strategies, the instance name).
A body with the actual data.

Example

@Value
class Signal {

  String name;
  int value;
}

@ExportStrategy
class ExampleStrategy {

  private MessageChannel<Signal> signalChannel;

  @Override
  public void onStarted() {
    this.signalChannel = this.context.openChannel(Signal.class, "test-signal-channel");
  }

  @Override
  public void onMessage(MessageChannel<?> channel) {
    // In case multiple channels are used, check for the exact channel object
    if (this.signalChannel == channel) {
      final Message<Signal> message = this.signalChannel.getMessage();
      final Signal signal = message.getBody();
      log.info("received signal message from {} (time: {}): name={}, value={}",
          message.getOrigin(),
          message.getTime(),
          signal.getName(),
          signal.getValue());
    }
  }

  @Override
  public void onHeartbeat() {
    this.signalChannel.sendMessage(new Signal("foo", 123));
  }
}

Access from external applications

External applications can use channels to:

Push custom data feeds or signals into the trading system.
Interact with running strategies.
Process analytical data published by strategies.

Tip

When many strategies publish on a shared channel and only a single external receiver reads it, have the producers pause() the channel. A paused channel can still publish but doesn't receive, which avoids needless work.

@Override
public void onStarted() {
  this.channel = this.context.openChannel(String.class, "channel");
  this.channel.pause();
}

Kafka integration

Channels can be wired to Kafka topics, so that messages flow into or out of Kafka.

Requirements:

The Strategy Server is configured with the Kafka cluster's connection details and a channel-to-topic mapping.
Channels writing to Kafka use either KafkaMessage or String as their data type.
Channels reading from Kafka always use KafkaMessage.

Note

Kafka records are key/value pairs (the key drives partition distribution). Both key and value are specified in a KafkaMessage.

For messages originating from String channels, the record key is either:

The process ID, or
null (if kafka_use_process_id_as_string_key is false).

Configuring Kafka

The Strategy Server configuration controls the Kafka connection. The relevant options:

Option	Description
`kafka_servers`	A list containing all the brokers of the Kafka cluster.
`kafka_stop_on_error`	If true, strategies will be stopped if writing to Kafka fails.
`kafka_topics`	A list of topics mapped to message channels. Each channel has the same name as the topic. Messages published on these channels are automatically written to the corresponding Kafka topic.

Topic permissions:

r → read-only
w → write-only (default)
rw → read/write

{
  "kafka_servers": [
    "broker1:9092",
    "broker2:9092"
  ],
  "kafka_stop_on_error": true,
  "kafka_topics": [
    "readOnlyTopic1:r",
    "readOnlyTopic2:r",
    "writeOnlyTopic:w",
    "readWriteTopic:rw",
    "defaultWriteOnlyTopic"
  ]
}

Kafka record headers

Every record written to Kafka carries:

Header	Description
`CS_ORIGIN`	Origin of the message, either the strategy instance name or the WebSocket/REST API client name.
`CS_PROCESS_ID`	Process ID of the strategy server.

`KafkaMessage` class

A strategy writes to Kafka by opening the channel as KafkaMessage and sending values:

@Override
public void onStarted() {
  this.channel = this.context.openChannel(KafkaMessage.class, "mytopic1");
  this.channel.pause();
  this.channel.sendMessage(KafkaMessage.of("key", "value"));
}

A KafkaMessage is a key-value pair. Both fields are byte buffers. The class has static helpers for common key/value types like String and long.

Tip

Depending on how the Kafka cluster is configured (AVRO, JSON, XML), you may need to construct the byte buffers yourself.

Message Channels