## 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

+++ Internal Channels (Strategy-to-Strategy)

- 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.

+++ External Channels (HTTP/WebSocket Access)

- Data type must be `String`.
- Strategies serialise on send and deserialise on receive.
- Reachable from outside via REST or WebSocket
  (see [Strategy Server API](/strategy-execution/strategy-server-api.md)).

+++

### Basic usage

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

```java
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

```java

@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.

```java

@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:

1. The Strategy Server is configured with the Kafka cluster's connection
   details and a channel-to-topic mapping.
2. Channels writing to Kafka use either `KafkaMessage` or `String` as
   their data type.
3. 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

```json
{
  "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:

```java

@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.

---