1.0.0-M2

1.0.0-M2 is the last milestone allowed to break the API before 1.0.0. This page lists the breaking changes of the restructuring of the SPI (the runtime machinery is now hidden in pipelite-core, and an adapter author sees only the contracts it implements), for people already using 1.0.0-M1. The new features of the milestone are at the end.

Application code

queue-channel-adapter is gone. queue:// is now built into pipelite-core. Remove the queue-channel-adapter dependency if you declared it. Flows that use queue:// need no change.
default-channel-adapters is gone. It only grouped time and slf4j, which pipelite-core now depends on directly: remove the default-channel-adapters dependency if you declared it (declare time-channel-adapter or slf4j-channel-adapter only if you use them without pipelite-core).
PipeliteContext shows less. It keeps registering flows and dependencies, addChannelConfigurer, isRegistered, getFlowDefinition, start and stop. tryFindFlow, tryFindFlowByName, getEndpointFactory, getExchangeFactory, getAggregateRepository, getDurableInboxProvider and getSourceWorkerPool are no longer part of it: they were the framework’s own plumbing.
Creating and supplying an exchange. To hand an exchange to a destination, create it with context.createExchange(payload) and give it to context.supplyExchange(url, exchange), which now takes the Exchange interface. getExchangeFactory().createExchange(…) is gone.
The values the framework keeps on an exchange are no longer headers. The return address, the routing slip and the details of a failure (X-Failure-Exception-Type, X-Failure-Exception-Message, X-Failure-Stack-Trace) are exchange properties. Code that read them with tryGetHeader(…) must not rely on them: they are not part of the public API.
Moved types. SourceConcurrencyConfigurer and ExecutorType moved from io.pipelite.spi to io.pipelite.dsl.definition. Update the imports of code that configures the concurrency of a queue source.
A flow cannot read from a sink-only channel. fromSource("slf4j://…") used to start and then receive nothing; it now fails when the context starts.
HTTP: a request whose exchange cannot be accepted is answered with 500.

Channel adapter authors

The contract an adapter implements is smaller, and everything around it belongs to the framework. See Custom Adapter for the whole of it.

In 1.0.0-M1 Now

In `1.0.0-M1`	Now
`Endpoint.createConsumer()`, returning a consumer built on `EventDrivenConsumerService` or `DefaultPollingConsumer`	`Endpoint.createSource()`, returning an `EventSource` (pushes through `SourceContext.accept`) or a `PollingSource` (`poll()`). The consumers, their queue, workers and backpressure are the framework’s.
`Producer` built on `DefaultProducer` / `AbstractProducer`, with `doProcess(ExchangeImpl)`	`Producer` is an interface: `produce(Exchange)`, plus an optional `close()` that the framework calls.
`Endpoint extends DefaultEndpoint`	`DefaultEndpoint` is removed: implement `Endpoint` (`getEndpointURL()`, and the side or sides you support).
`ExchangeImpl`, `ExchangeFactory`, `ExchangeFactoryAware`	Not visible to an adapter. A source creates exchanges with `SourceContext.newExchange(payload)`; a producer receives the `Exchange` interface.
`DurableInbox` and the other inbox types	Not visible: the framework writes to the inbox before it lets a source acknowledge its input.
`ContextEventListener.onFlowRegistered(Flow)`	Removed from the SPI; `onContextStarted` and `onContextStopped` remain.

Endpoint.createConsumer(), returning a consumer built on EventDrivenConsumerService or DefaultPollingConsumer

Endpoint.createSource(), returning an EventSource (pushes through SourceContext.accept) or a PollingSource (poll()). The consumers, their queue, workers and backpressure are the framework’s.

Producer built on DefaultProducer / AbstractProducer, with doProcess(ExchangeImpl)

Producer is an interface: produce(Exchange), plus an optional close() that the framework calls.

Endpoint extends DefaultEndpoint

DefaultEndpoint is removed: implement Endpoint (getEndpointURL(), and the side or sides you support).

ExchangeImpl, ExchangeFactory, ExchangeFactoryAware

Not visible to an adapter. A source creates exchanges with SourceContext.newExchange(payload); a producer receives the Exchange interface.

DurableInbox and the other inbox types

Not visible: the framework writes to the inbox before it lets a source acknowledge its input.

ContextEventListener.onFlowRegistered(Flow)

Removed from the SPI; onContextStarted and onContextStopped remain.

A source acknowledges its input (commits an offset, answers a request) only after SourceContext.accept returned, and must not acknowledge it when accept throws.

pipelite-spi now contains only the types an adapter author sees and depends on pipelite-dsl alone. Adapter unit tests can no longer build the framework’s Exchange implementation: use a stub of the Exchange interface, or add pipelite-core as a test dependency. The Exchange interface has a new method, getIdempotencyKey(): a stub of it must implement it. SourceContext has a new method, newExchange(payload, transportHeaders), that an adapter with headers calls with all of them (a stub of the context must implement it), and Producer a method produce(exchange, headers) that the framework calls with the headers the flow chose to send and that an adapter with headers overrides.

Build

A bill of materials. pipelite-bom manages the version of every published module: import it once and declare the artifacts without a version (see Installation).

New

Barrier. A flow can make an exchange wait until other flows have completed their work for the same business key, with .barrier(…). It is a new step: existing flows are not affected. See Barrier.
Idempotency key. Exchange.getIdempotencyKey() is a key that identifies the piece of work an exchange carries and is the same in every step, retry, restart and flow, to hand to the external systems a step calls. A source can declare it from the request with idempotencyKey(…). See Idempotency key.
Headers from and to the outside. A source puts on the exchange the headers of the request or of the record, and a sink sends the headers of the exchange to the other system, for the ones the flow chooses with propagateHeaders(h → h.filter("Idempotency-Key")) (or h.all()), none by default. It is the same for every channel adapter that has headers: propagateHeaders is a capability (PropagatesHeaders) of the settings of the endpoints that have them, so it is not offered on a file, a timer, a log or a queue. A sink now has typed settings too, toSink(url, (KafkaSinkConfigurer c) → …), and an adapter offers them with ChannelAdapter#newSinkConfigurer(). The http:// source also offers the method and the path of the request, and each parameter of the query (Http-Query-<name>), and the kafka:// source the topic, the partition, the offset and the timestamp of the record. See Headers from and to the outside.
Ordered concurrency. A queue source with concurrency can keep the exchanges that have the same key in the order they arrived, while the others run in parallel, with orderedBy(…). Nothing changes for a source that does not declare it. See Order and concurrency.