Function ws_encode_task

crates/client-api/src/routes/subscribe.rs:1430–1455

Task that reads [`OutboundWsMessage`]s from `messages`, encodes them, and sends the resulting [`Frame`]s to `outgoing_frames`. Meant to be [`tokio::spawn`]ed. The function also takes care of reusing serialization buffers and reporting metrics via [`SendMetrics`].

(
    metrics: SendMetrics,
    config: ClientConfig,
    mut messages: mpsc::UnboundedReceiver<OutboundWsMessage>,
    outgoing_frames: mpsc::UnboundedSender<Frame>,
    bsatn_rlb_pool: BsatnRowListB

/// The function also takes care of reusing serialization buffers and reporting
/// metrics via [`SendMetrics`].
async fn ws_encode_task(
    metrics: SendMetrics,
    config: ClientConfig,
    mut messages: mpsc::UnboundedReceiver<OutboundWsMessage>,
    outgoing_frames: mpsc::UnboundedSender<Frame>,
    bsatn_rlb_pool: BsatnRowListBuilderPool,
) {
    let mut encoder = WsEncoder {
        config,
        buffers: SerializeBufferPool::new(config),
        metrics: &metrics,
        outgoing_frames: &outgoing_frames,
        bsatn_rlb_pool: &bsatn_rlb_pool,
        binary_server_messages: Vec::new(),
    };
    let mut message_batch = Vec::new();
    while messages.recv_many(&mut message_batch, ENCODE_BATCH_SIZE).await != 0 {
        log::trace!("encoding batch of {} websocket messages", message_batch.len());
        // `encode_batch` drains `message_batch` on success. If forwarding to
        // the websocket send loop fails, the receiver is gone, so the encode
        // task can terminate.
        if encoder.encode_batch(&mut message_batch).await.is_err() {
            break;
        }
    }
}

/// Stateful websocket encoder for one client connection.
///