AblyDocs

Adaptor and decoder

Internal adaptor and decoder components: the pipe writer that streams and repairs appends, and delivery through the codec.

An agent adaptor is one channel and one codec, and a decoder reads the messages of the same codec. These are the parts inside them, and how they coordinate the codec, the channel and your application code.

Pipe writer

pipe() reads a source of events one at a time, encodes each one, and gives the messages to a pipe writer. The writer keeps a key table for the life of the pipe. A codec names streams by its own keys, such as a message id or a tool call id, because it cannot know the serial Ably returns for a publish:

  • publish under a key publishes the message and remembers its serial.
  • append to a key that is not live publishes the message, waits for its serial and remembers it, so a stream's first delta opens the message. append to a live key appends.
  • update replaces the content of the message under a live key, and ends forgets a key.

The writer sends appends without waiting for them, so a response streams at the speed of the connection rather than one round trip per token. It waits for a stream's pending appends before it sends the message that ends the stream. Ably rolls up the appends a connection publishes within a short window into one delivery, and the wait keeps the end message from being delivered ahead of the stream's last text.

If an append fails, the writer carries on. When the stream ends, it repairs the message with one update that carries the full content, before the end message. History then holds the whole message, and a live client reads the missing text as the update's new part. If the repair fails, the pipe rejects with PipeFailed.

On every exit, cancellation included, the pipe flushes and repairs what it wrote and releases the source before it settles, so a cancelled pipe leaves whole messages on the channel.

Delivery

Your application typically subscribes the decoder with channel.subscribe(decoder.receive), which adds the listener and then attaches the channel, so no message can arrive before something receives it. Every inbound message goes through the codec:

  • Each event the codec decodes becomes one delivery, with the event and the Ably message.
  • A message the codec has nothing for, such as another publisher's message, a message the codec has already seen, or a deleted message, becomes one delivery with event set to undefined.
  • A message whose decode throws becomes one delivery with event set to undefined, and the error goes to the decoder's onError handlers as SessionMessageProcessingFailed.

The application always sees the raw message and decides what to do with it. Every handler receives every delivery. A handler that throws is logged, and the other handlers still receive the delivery.

History replay

Your application can read history itself and pass each message to decoder.decode, which returns the deliveries without emitting them. Use the same decoder for history and live messages, so a message that both carry decodes once while its stream is open. History pages arrive newest first, and the decoder does not reorder deliveries. To receive history and live messages in order from one subscription, your application can rewind the channel to a serial stored in your database.

Channel continuity

The adaptor and the decoder do not watch the channel's state. Your application should watch the channel's state to detect a loss of continuity, and then recover the conversation.

The handler still receives the full text of a stream in flight during the gap. The first message Ably delivers after the channel attaches again carries the full content so far, and the decoder core passes on only the part the handler has not received.

Close

A pipe stops when the signal passed to pipe() fires, and flushes and repairs what it wrote before it rejects with OperationCancelled. To stop reading, your application can call channel.unsubscribe(decoder.receive). The Ably client stays connected, because your application owns it.

Related pages