Exchange

The Exchange is the fundamental vehicle through which data travels along a flow. Each time a source produces an event, a new Exchange is created and passed through all nodes in the pipeline.

Two types, one exchange

What your code sees is the io.pipelite.dsl.Exchange interface: the input and output payload, and the headers. It is what Processor, ExceptionHandler and Condition implementations receive.

The framework, and the authors of channel adapters, work with io.pipelite.spi.flow.exchange.ExchangeImpl, the concrete class behind it. It adds the message containers, the internal properties and forwardIfNecessary(), which is why the channel adapter SPI (Producer, Consumer, Flow#supply) is written in terms of ExchangeImpl.

Structure

public class ExchangeImpl implements Exchange, Serializable {
    private final Message input;    // incoming message (immutable after creation)
    private Message output;         // response message (optional)
    private final Headers headers;  // cross-cutting metadata
    private final Map<String, Object> properties; // internal routing state
}

Payload access

// Read the incoming payload
Object raw = exchange.getInputPayload();
MyDto dto  = exchange.getInputPayloadAs(MyDto.class);

// Check the payload type
Class<?> type = exchange.getInputPayloadType();

// Set the output payload
exchange.setOutputPayload(result);

// Automatic forward (input → output if output is empty)
exchange.forwardIfNecessary();

Headers

Headers are string key-value pairs shared across all nodes in the flow. The methods below are shortcuts to exchange.getHeaders(), which returns the underlying Headers when you need it (the expression engine uses it to resolve Headers['name']).

// Write
exchange.putHeader("correlation-id", "abc123");

// Safe read
Optional<String> id = exchange.tryGetHeader("correlation-id");

// Read with expectation (throws exception if absent)
String id = exchange.expectHeader("correlation-id");

// Typed read
Optional<Integer> count = exchange.tryGetHeaderAs("retry-count", Integer.class);

// Removal
exchange.removeHeader("temp-header");

Properties

Properties are Object key-value pairs used internally by the framework for routing.

exchange.setProperty("my-key", myObject);
MyType value = exchange.getProperty("my-key", MyType.class);
MyType valueOrDefault = exchange.getPropertyOrDefault("my-key", MyType.class, defaultValue);
exchange.removeProperty("my-key");