BooblikConnection

What this module is responsible for

The BooblikConnection class is responsible for managing a single TCP connection to a Booblik broker. It implements a pipelined communication protocol where multiple requests can be in flight simultaneously. It handles the encoding of requests, the asynchronous reading of responses, and the complex task of matching those responses back to the specific coroutines that requested them, all while maintaining strict FIFO ordering guarantees.

Diagram

Pipelined Request-Response Matching

Because the broker is guaranteed to answer requests in the exact order they were received, the client uses a FIFO queue to manage pending responses. When a request is sent, a Pending object is added to a ConcurrentLinkedQueue (as seen in BooblikConnection.kt:60). When a response arrives, the reader coroutine polls the head of this queue and validates that the correlationId in the response matches the expected ID via checkOrder (as seen in BooblikConnection.kt:193). If the IDs do not match, it indicates the broker has violated the protocol, and the client fails loudly.

The Outbound Writer Coroutine

To prevent interleaved writes where bytes from two different requests might mix on the wire, BooblikConnection uses a single-writer pattern. All callers send their requests to an outbound Channel (as seen in BooblikConnection.kt:59), and a single dedicated writer coroutine drains this channel to perform the actual socket writes (as seen in BooblikConnection.kt:73). This ensures that the order in which requests are enqueued in the channel is exactly the order in which they are written to the SocketChannel.

AckPolicy.NONE and Fire-and-Forget Semantics

The client supports a "fire-and-forget" mode through the AckPolicy.NONE setting. When a request is produced with this policy, the produce function skips the creation of a CompletableDeferred and does not add anything to the pending queue (as seen in BooblikConnection.kt:110). The request is simply sent to the outbound channel, and the function returns null immediately, bypassing the response matching logic entirely.

ConnectionClosedException and Failure Propagation

When a connection fails or is closed, the client must ensure that no caller is left hanging forever. The fail function handles this by catching the error and iterating through the pending queue to fail every waiting request with a ConnectionClosedException (as seen in BooblikConnection.kt:166). This ensures that all suspended await() calls in the calling coroutines are resumed with an exception, maintaining the integrity of the asynchronous flow.

Correlation ID Atomicity

To support high concurrency, BooblikConnection uses an AtomicInteger to generate correlationIds (as seen in BooblikConnection.kt:71). This is critical because multiple coroutines call produce, metadata, or fetch concurrently. Using getAndIncrement() ensures that every single request receives a unique, monotonically increasing ID, preventing the catastrophic failure where two different callers receive each other's responses (as noted in the documentation at BooblikConnection.kt:67).

Key files

Behaviour that surprise

• Strict Ordering Requirement: If the broker sends a response out of order, the client will throw an error rather than attempting to find the correct caller, because the protocol's FIFO guarantee is a prerequisite for correctness (as seen in BooblikConnection.kt:34).
• Blocking the Pipeline: A single fetch request with a long maxWaitMillis will block all subsequent requests on the same connection, including produce requests, because the connection is served strictly one request at a time (as seen in BooblikConnection.kt:132).
• Silent Failures in Acceptor: In the server implementation, if the acceptor loop encounters an exception, it is treated as transient to prevent the broker from entering a state where it accepts TCP connections but never processes them (as seen in BooblikServer.kt:169).