Redis (RESP) Protocol Layer — Shape 2 Design

[SeanTAllen]

Design for a Redis protocol layer using the "protocol class as lifecycle interceptor" pattern (Shape 2 from Discussion ponylang/lori#187).

Design overview

The Redis protocol library provides two main classes — RedisClient and RedisServer — that sit between TCPConnection and the user. Each class implements a Lori lifecycle receiver trait, intercepting raw byte callbacks and delivering typed Redis protocol events to the user via a callbacks trait. The user's actor implements TCPConnectionActor (for ASIO plumbing) and the protocol callbacks trait. The user never touches raw bytes or Lori lifecycle callbacks directly.

A convenience trait (RedisClientActor / RedisServerActor) eliminates _connection() boilerplate by providing a default implementation that delegates through the protocol class.

RESP data types

type RedisValue is
  ( RedisSimpleString
  | RedisError
  | RedisInteger
  | RedisBulkString
  | RedisArray
  | RedisNull )

class val RedisSimpleString
  let value: String val
  new val create(value': String val) => value = value'

class val RedisError
  let message: String val
  new val create(message': String val) => message = message'

class val RedisInteger
  let value: I64
  new val create(value': I64) => value = value'

class val RedisBulkString
  let value: Array[U8] val
  new val create(value': Array[U8] val) => value = value'

class val RedisArray
  let values: Array[RedisValue] val
  new val create(values': Array[RedisValue] val) => values = values'

primitive RedisNull

All types are val so they can be stored, compared, and passed freely within the actor. Even though we're in a single-actor model (Shape 2), making them val costs nothing for small values and allows the types to be used in val collections (e.g., a response cache). The parser builds values as iso in recover blocks and consumes them to val.

RedisError here is a RESP error response (the -ERR ... wire type), not a protocol layer error. Protocol layer errors are separate types (see below).

Send error types

primitive RedisSendNotConnected
  """
  The connection is not established, has been closed, or the Redis protocol
  is not ready (e.g., SSL handshake in progress, or in pub/sub mode where
  the command is not allowed).
  """

primitive RedisSendNotWriteable
  """
  The socket is under backpressure. Wait for `_on_redis_unthrottled` before
  retrying.
  """

primitive RedisSendInvalidMode
  """
  The command is not valid in the current mode. For example, calling
  `command()` while in pub/sub mode (use `subscribe()`/`unsubscribe()`
  instead), or calling `subscribe()` while in normal mode and there are
  pending pipelined commands whose responses haven't arrived yet.
  """

type RedisSendError is
  ( RedisSendNotConnected
  | RedisSendNotWriteable
  | RedisSendInvalidMode )

Preserves the transient/permanent distinction from Lori's SendErrorNotWriteable (transient — retry after unthrottle) vs SendErrorNotConnected (permanent). Adds RedisSendInvalidMode for protocol-level state machine violations.

Protocol error type

primitive RedisProtocolError
  """
  The RESP parser encountered malformed data: invalid type byte, negative
  bulk string length (other than -1 for null), non-numeric characters in a
  length or integer field, or missing CRLF terminator. The byte stream is
  likely desynchronized and cannot be recovered.
  """

Request token

class val RedisRequestToken is Equatable[RedisRequestToken]
  """
  Identifies a Redis command sent via `command()`, `subscribe()`, or
  `unsubscribe()`. Delivered to `_on_redis_response` when Redis replies.

  Independent of Lori's SendToken — a SendToken tracks "bytes reached the
  OS" while a RedisRequestToken tracks "Redis processed the command." The
  protocol class correlates them internally; the user never sees SendToken.
  """
  let _id: USize

  new val _create(id': USize) =>
    _id = id'

  fun eq(that: box->RedisRequestToken): Bool =>
    _id == that._id

  fun ne(that: box->RedisRequestToken): Bool =>
    not eq(that)

RedisClient — client-side protocol class

class RedisClient is ClientLifecycleEventReceiver
  """
  Lifecycle interceptor for client-side Redis connections. Implements
  `ClientLifecycleEventReceiver` so that Lori delivers raw byte callbacks
  here instead of to the user's actor. Parses RESP, manages the protocol
  state machine (normal mode, pub/sub mode), tracks pipelined requests,
  and delivers typed Redis events to the user via `RedisClientCallbacks`.

  Created inside the user's actor constructor. The user's actor is the
  `TCPConnectionActor` (ASIO plumbing); this class is the lifecycle
  receiver (application callbacks). Both are passed to TCPConnection as
  separate parameters.

  The class also handles SendToken correlation: it receives `_on_sent` and
  `_on_send_failed` from Lori and translates them into protocol-level
  events, hiding SendToken from the user entirely.
  """

  // --- Lori connection ---
  var _tcp_connection: TCPConnection = TCPConnection.none()

  // --- Protocol state ---
  // The parse buffer is ref, not iso — it lives inside the protocol class
  // (which is ref), never crosses actor boundaries, and needs to be
  // freely readable and writable during parsing. Using iso here would
  // require swap tricks at every access point for no benefit.
  let _parse_buffer: Array[U8] = Array[U8]
  var _parse_offset: USize = 0
  var _mode: RedisMode = RedisNormalMode
  var _next_token_id: USize = 0
  // Pending request queue for pipelining. Responses arrive in order,
  // so we dequeue from the front as responses arrive.
  let _pending_requests: Array[RedisRequestToken] = Array[RedisRequestToken]
  // Maps SendToken.id to RedisRequestToken for _on_sent/_on_send_failed
  // correlation. Entries are added in command(), removed in _on_sent/
  // _on_send_failed.
  let _send_token_map: Map[USize, RedisRequestToken] =
    Map[USize, RedisRequestToken]

  // --- Pub/sub tracking ---
  // Queue of expected confirmation counts, one entry per subscribe() or
  // unsubscribe() call. A single subscribe(["a", "b"]) pushes 2 onto
  // this queue because Redis sends 2 confirmations (one per channel).
  // As confirmations arrive, we decrement the front entry. When it reaches
  // zero, we dequeue the matching RedisRequestToken from _pending_requests
  // and pop this entry.
  //
  // A value of 0 at the front is a sentinel meaning "consume until the
  // subscription count in the confirmation reaches zero" — used for
  // unsubscribe-all (unsubscribe([])) where we don't know the count.
  let _pending_sub_confirmations: Array[USize] = Array[USize]

  // --- Callbacks ---
  // Union with None for the none() constructor — Pony primitives are val,
  // which can't satisfy a ref field. Matching Lori's own pattern
  // (TCPConnection stores lifecycle_event_receiver as union with None).
  let _callbacks: (RedisClientCallbacks ref | None)

  // ============================================================
  // Constructors
  // ============================================================

  new create(auth: TCPConnectAuth,
    host: String,
    port: String,
    enclosing: TCPConnectionActor ref,
    callbacks: RedisClientCallbacks ref)
  =>
    """
    Create a plaintext Redis client connection. `_on_redis_connected` fires
    when the TCP connection is established and the client is ready to send
    commands.
    """
    _callbacks = callbacks
    _tcp_connection = TCPConnection.client(auth, host, port, "", enclosing,
      this)

  new ssl(auth: TCPConnectAuth,
    ssl_ctx: SSLContext val,
    host: String,
    port: String,
    enclosing: TCPConnectionActor ref,
    callbacks: RedisClientCallbacks ref)
  =>
    """
    Create a TLS-encrypted Redis client connection. `_on_redis_connected`
    fires after both the TCP connection and TLS handshake complete. If the
    TLS handshake fails, `_on_redis_connection_failure` fires instead.
    """
    _callbacks = callbacks
    _tcp_connection = TCPConnection.ssl_client(auth, ssl_ctx, host, port, "",
      enclosing, this)

  new none() =>
    """
    Placeholder for field initialization before the real constructor runs.
    See the `none()` discussion below for why this may be unnecessary.
    """
    _callbacks = None

  // ============================================================
  // Public API
  // ============================================================

  fun ref connection(): TCPConnection =>
    """
    Returns the underlying TCPConnection. Used by the user's actor to
    satisfy TCPConnectionActor._connection(). Also provides access to
    `keepalive()`, `local_address()`, `remote_address()`, `mute()`,
    `unmute()`, `close()`.
    """
    _tcp_connection

  fun ref command(args: Array[ByteSeq] val):
    (RedisRequestToken | RedisSendError)
  =>
    """
    Send a Redis command. Returns a token on success, or an error
    explaining the failure.

    In normal mode: serializes `args` to RESP array-of-bulk-strings
    format, sends via the TCP connection, tracks the token for pipelining.
    The matching response arrives via `_on_redis_response(token, value)`.

    In pub/sub mode: returns `RedisSendInvalidMode`. Use `subscribe()`
    and `unsubscribe()` instead.

    If the connection is not open or not writeable, returns the appropriate
    `RedisSendNotConnected` or `RedisSendNotWriteable`.
    """
    if _mode isnt RedisNormalMode then
      return RedisSendInvalidMode
    end
    _send_command(args)

  fun ref subscribe(channels: Array[String] val):
    (RedisRequestToken | RedisSendError)
  =>
    """
    Subscribe to one or more channels. Transitions the connection to
    pub/sub mode (if not already there). Returns a token; a single
    `_on_redis_response` callback fires after all subscription
    confirmations for this call have arrived (one per channel).
    The delivered value is the final confirmation (a RESP array:
    ["subscribe", channel, count]).

    Allowed in both normal and pub/sub mode. In normal mode, cannot be
    called while there are pending pipelined responses (returns
    `RedisSendInvalidMode`) because entering pub/sub mode with
    outstanding normal-mode responses would desynchronize the response
    queue.

    Push messages for subscribed channels arrive via
    `_on_redis_push(channel, message)`.
    """
    if (_mode is RedisNormalMode) and (_pending_requests.size() > 0) then
      return RedisSendInvalidMode
    end
    let args: Array[ByteSeq] val = recover
      let a = Array[ByteSeq](channels.size() + 1)
      a.push("SUBSCRIBE")
      for ch in channels.values() do a.push(ch) end
      a
    end
    let result = _send_command(args)
    match result
    | let _: RedisRequestToken =>
      // Only transition to pub/sub mode after confirming the send
      // succeeded. If the send fails, the mode stays as-is and the
      // caller gets a RedisSendError.
      _mode = RedisPubSubMode
      // Redis sends one confirmation per channel. Push the expected count
      // onto the queue so we can correlate all confirmations back to this
      // single token.
      _pending_sub_confirmations.push(channels.size())
    end
    result

  fun ref unsubscribe(channels: Array[String] val):
    (RedisRequestToken | RedisSendError)
  =>
    """
    Unsubscribe from one or more channels. If `channels` is empty,
    unsubscribes from all channels. When the subscription count reaches
    zero, the connection transitions back to normal mode.

    Only valid in pub/sub mode. Returns `RedisSendInvalidMode` if called
    in normal mode.
    """
    if _mode isnt RedisPubSubMode then
      return RedisSendInvalidMode
    end
    let args: Array[ByteSeq] val = recover
      let a = Array[ByteSeq](channels.size() + 1)
      a.push("UNSUBSCRIBE")
      for ch in channels.values() do a.push(ch) end
      a
    end
    // For UNSUBSCRIBE with explicit channels, Redis sends one confirmation
    // per channel. For UNSUBSCRIBE with no channels (unsubscribe all),
    // Redis sends one confirmation per currently subscribed channel — but
    // we don't know the count here. Push 0 as a sentinel meaning "consume
    // until subscription count reaches zero."
    let result = _send_command(args)
    match result
    | let _: RedisRequestToken =>
      _pending_sub_confirmations.push(channels.size())
      // channels.size() is 0 for unsubscribe-all, which serves as the
      // sentinel — see _dispatch_unsubscription_confirmation.
    end
    result

  fun mode(): RedisMode =>
    """
    Returns the current protocol mode (normal or pub/sub).
    """
    _mode

  // ============================================================
  // ClientLifecycleEventReceiver implementation (intercepted)
  // ============================================================

  fun ref _connection(): TCPConnection =>
    """
    Required by ClientLifecycleEventReceiver. Returns the same
    TCPConnection that the user's _connection() returns (via
    connection()). Both methods must return the same object — this is a
    correctness constraint of the interceptor pattern. See Discussion ponylang/lori#187
    for details on this dual-_connection() requirement.
    """
    _tcp_connection

  fun ref _on_connecting(inflight_connections: U32) =>
    // Absorbed. Happy Eyeballs connection count is TCP-level detail
    // irrelevant to Redis protocol operation.
    None

  fun ref _on_connected() =>
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_connected()
    end

  fun ref _on_connection_failure() =>
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_connection_failure()
    end

  fun ref _on_closed() =>
    _pending_requests.clear()
    _pending_sub_confirmations.clear()
    _mode = RedisNormalMode
    _parse_buffer.clear()
    _parse_offset = 0
    // Do NOT clear _send_token_map here. Lori's _on_send_failed fires
    // AFTER _on_closed (in a subsequent behavior turn). Clearing the map
    // here would prevent _on_redis_send_failed from ever reaching the user.
    // Entries are cleaned up by _on_send_failed, or become garbage when
    // the object is collected.
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_closed()
    end

  fun ref _on_received(data: Array[U8] iso) =>
    _feed(consume data)

  fun ref _on_throttled() =>
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_throttled()
    end

  fun ref _on_unthrottled() =>
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_unthrottled()
    end

  fun ref _on_sent(token: SendToken) =>
    // Correlate SendToken → RedisRequestToken and deliver protocol-level
    // send confirmation. The user never sees SendToken.
    try
      let redis_token = _send_token_map(token.id)?
      _send_token_map.remove(token.id)?
      match _callbacks
      | let cb: RedisClientCallbacks ref => cb._on_redis_sent(redis_token)
      end
    end

  fun ref _on_send_failed(token: SendToken) =>
    // A send was accepted but the bytes couldn't be delivered (connection
    // closed with a partial write pending). Translate to protocol level.
    try
      let redis_token = _send_token_map(token.id)?
      _send_token_map.remove(token.id)?
      match _callbacks
      | let cb: RedisClientCallbacks ref => cb._on_redis_send_failed(redis_token)
      end
    end

  fun ref _on_tls_ready() =>
    // TLS upgrade via start_tls() completed. This would only happen if
    // the user called start_tls() directly on the TCPConnection (bypassing
    // the protocol layer). Since Redis TLS is typically established via the
    // ssl constructor, this callback is unusual. Forward it as a connected
    // event.
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_connected()
    end

  fun ref _on_tls_failure() =>
    match _callbacks
    | let cb: RedisClientCallbacks ref => cb._on_redis_connection_failure()
    end

  // ============================================================
  // Internal: RESP parsing
  // ============================================================

  fun ref _feed(data: Array[U8] iso) =>
    """
    Append data to the parse buffer and extract all complete RESP values.
    For each complete value, dispatch to the appropriate callback based on
    the current mode and pending request queue.
    """
    // Append new data to parse buffer. The incoming data is iso (from
    // Lori's _on_received), consumed into the ref parse buffer.
    let incoming: Array[U8] val = consume data
    _parse_buffer.append(incoming)

    // Parse loop: extract complete RESP values until we can't
    while true do
      match _try_parse()
      | let value: RedisValue =>
        _dispatch_value(value)
      | _RespIncomplete =>
        // Need more data. Compact the buffer if we've consumed a
        // significant prefix.
        _compact_buffer()
        break
      | _RespParseError =>
        // Malformed RESP data. The byte stream is desynchronized.
        _protocol_error_and_close()
        break
      end
    end

  fun ref _try_parse(): (_RespParseResult | RedisValue) =>
    """
    Attempt to parse one complete RESP value starting at _parse_offset.
    Returns the parsed value if complete, _RespIncomplete if more data is
    needed, or _RespParseError if the data is malformed.

    Handles recursive parsing for arrays (an array element can be any RESP
    type, including nested arrays).
    """
    // Implementation details:
    // - Read the type byte at _parse_offset
    // - For +, -, :: scan for \r\n, extract the line
    // - For $: parse length, check for null (-1), then read exactly
    //   length bytes + \r\n
    // - For *: parse count, check for null (-1), then recursively parse
    //   count elements
    // - Advance _parse_offset past the consumed bytes
    // - If any read would go past _parse_buffer.size(), return
    //   _RespIncomplete
    _RespIncomplete // placeholder

  fun ref _compact_buffer() =>
    """
    If _parse_offset has advanced past a significant portion of the buffer,
    shift the remaining bytes to the front to prevent unbounded growth.

    Uses remove() to drop the consumed prefix. Since the buffer is ref
    (not iso), we can operate on it directly.
    """
    if _parse_offset > 4096 then
      _parse_buffer.remove(0, _parse_offset)
      _parse_offset = 0
    end

  fun ref _dispatch_value(value: RedisValue) =>
    """
    Route a parsed RESP value to the appropriate callback based on
    the current mode.
    """
    match _mode
    | RedisNormalMode =>
      // In normal mode, responses correspond to pending requests in order.
      if _pending_requests.size() == 0 then
        // More responses than requests — protocol desynchronization.
        _protocol_error_and_close()
        return
      end
      try
        let token = _pending_requests.shift()?
        match _callbacks
        | let cb: RedisClientCallbacks ref => cb._on_redis_response(token, value)
        end
      end
    | RedisPubSubMode =>
      // In pub/sub mode, messages are either:
      // 1. Subscription confirmations: ["subscribe", channel, count]
      //    or ["unsubscribe", channel, count]
      // 2. Push messages: ["message", channel, payload]
      // 3. Error responses (e.g., "-ERR ...") — Redis can send errors
      //    in pub/sub mode for invalid commands or server errors.
      //
      // Subscription confirmations are tracked by _pending_sub_confirmations
      // (not _pending_requests) because a single subscribe(["a", "b"])
      // generates one RedisRequestToken but Redis sends two confirmations.
      // Push messages are delivered via _on_redis_push.
      // When unsubscribe count reaches 0, transition back to normal mode.
      match value
      | let arr: RedisArray =>
        if arr.values.size() < 3 then
          // All valid pub/sub messages have at least 3 elements
          // (type, channel, count/payload). A smaller array is malformed.
          _protocol_error_and_close()
          return
        end
        try
          match arr.values(0)?
          | let b: RedisBulkString =>
            let cmd = String.from_array(b.value)
            if (cmd == "message") or (cmd == "pmessage") then
              _dispatch_push_message(arr)
            elseif (cmd == "subscribe") or (cmd == "psubscribe") then
              _dispatch_subscription_confirmation(arr)
            elseif (cmd == "unsubscribe") or (cmd == "punsubscribe") then
              _dispatch_unsubscription_confirmation(arr)
            else
              // Unrecognized command type in pub/sub array
              _protocol_error_and_close()
            end
          else
            // Array's first element is not a bulk string
            _protocol_error_and_close()
          end
        end
      else
        // Any non-array RESP value in pub/sub mode is unexpected.
        _protocol_error_and_close()
      end
    end

  fun ref _dispatch_push_message(arr: RedisArray) =>
    """
    Route a pub/sub push message to _on_redis_push. Handles both formats:
    - "message":  ["message", channel, payload]
    - "pmessage": ["pmessage", pattern, channel, payload]

    For pmessage, the channel (not the pattern) is delivered as the
    channel argument to _on_redis_push. The pattern is not exposed in
    the current API — if needed, a separate _on_redis_pattern_push
    callback could be added.
    """
    // The caller (_dispatch_value) already validated arr.values.size() >= 3,
    // so index access for "message" elements won't fail on size. Type
    // mismatches (non-BulkString elements) are treated as protocol errors
    // since all valid pub/sub messages use bulk strings for channel/payload.
    try
      match arr.values(0)?
      | let cmd_bs: RedisBulkString =>
        let cmd = String.from_array(cmd_bs.value)
        if cmd == "message" then
          // ["message", channel, payload]
          match arr.values(1)?
          | let ch: RedisBulkString =>
            let channel = String.from_array(ch.value)
            match arr.values(2)?
            | let payload: RedisBulkString =>
              match _callbacks
              | let cb: RedisClientCallbacks ref =>
                cb._on_redis_push(channel, payload.value)
              end
            else
              _protocol_error_and_close()
            end
          else
            _protocol_error_and_close()
          end
        elseif cmd == "pmessage" then
          // ["pmessage", pattern, channel, payload] — requires 4 elements.
          // The caller validated >= 3 but pmessage needs >= 4.
          if arr.values.size() < 4 then
            _protocol_error_and_close()
            return
          end
          // Index 1 is pattern (skipped), index 2 is channel, index 3 is
          // payload.
          match arr.values(2)?
          | let ch: RedisBulkString =>
            let channel = String.from_array(ch.value)
            match arr.values(3)?
            | let payload: RedisBulkString =>
              match _callbacks
              | let cb: RedisClientCallbacks ref =>
                cb._on_redis_push(channel, payload.value)
              end
            else
              _protocol_error_and_close()
            end
          else
            _protocol_error_and_close()
          end
        end
      end
    end

  fun ref _protocol_error_and_close() =>
    """
    Deliver protocol error to callbacks and close the connection.
    Extracted to avoid repeating the match-on-callbacks pattern at every
    error site.
    """
    match _callbacks
    | let cb: RedisClientCallbacks ref =>
      cb._on_redis_protocol_error(RedisProtocolError)
    end
    _tcp_connection.hard_close()

  fun ref _consume_sub_confirmation(arr: RedisArray) =>
    """
    Shared logic for subscription and unsubscription confirmations.
    Decrements the front entry in _pending_sub_confirmations. When it
    reaches 1 (last expected confirmation), dequeues the matching
    RedisRequestToken from _pending_requests and delivers the final
    confirmation to the user.

    The sentinel value 0 (unsubscribe-all) is handled specially by the
    caller — _dispatch_unsubscription_confirmation bypasses this method's
    countdown and uses the subscription count from the confirmation to
    determine when to deliver the token.
    """
    // Separate try blocks so that queue desynchronization
    // (_pending_sub_confirmations has entries but _pending_requests is
    // empty) is detected as a protocol error rather than silently
    // swallowed.
    let remaining: USize =
      try _pending_sub_confirmations(0)?
      else return // No pending confirmations expected — ignore
      end

    if remaining == 0 then
      // Sentinel for unsubscribe-all. Don't decrement — the caller
      // (_dispatch_unsubscription_confirmation) handles delivery based
      // on the subscription count in the confirmation.
      return
    end

    if remaining == 1 then
      // Last confirmation for this subscribe/unsubscribe call.
      try _pending_sub_confirmations.shift()? end
      try
        let token = _pending_requests.shift()?
        match _callbacks
        | let cb: RedisClientCallbacks ref =>
          cb._on_redis_response(token, arr)
        end
      else
        // _pending_sub_confirmations had entries but _pending_requests
        // is empty — queue desynchronization, treat as protocol error.
        _protocol_error_and_close()
      end
    else
      // More confirmations expected for this call. Decrement and wait.
      try _pending_sub_confirmations(0)? = remaining - 1 end
    end

  fun ref _dispatch_subscription_confirmation(arr: RedisArray) =>
    """
    Route a subscribe/psubscribe confirmation. Delegates to
    _consume_sub_confirmation for the countdown logic.
    """
    _consume_sub_confirmation(arr)

  fun ref _dispatch_unsubscription_confirmation(arr: RedisArray) =>
    """
    Route an unsubscribe/punsubscribe confirmation. For explicit
    unsubscribe (known channel count), delegates to
    _consume_sub_confirmation for countdown logic. For unsubscribe-all
    (sentinel 0), waits for the subscription count in the confirmation
    to reach zero, then delivers the token.

    In both cases, checks whether the subscription count has reached
    zero to transition back to normal mode.
    """
    let remaining: USize =
      try _pending_sub_confirmations(0)?
      else return // No pending confirmations expected — ignore
      end

    if remaining == 0 then
      // Unsubscribe-all sentinel. Check the subscription count in the
      // confirmation to decide when to deliver the token and transition.
      //
      // Extract the count field. If it's missing or not a RedisInteger,
      // that's a malformed confirmation — protocol error.
      let count_value: I64 =
        try
          match arr.values(2)?
          | let count: RedisInteger => count.value
          else
            _protocol_error_and_close()
            return
          end
        else
          _protocol_error_and_close()
          return
        end

      if count_value == 0 then
        // All subscriptions are gone. Deliver the token and transition.
        try _pending_sub_confirmations.shift()? end
        try
          let token = _pending_requests.shift()?
          match _callbacks
          | let cb: RedisClientCallbacks ref =>
            cb._on_redis_response(token, arr)
          end
        else
          // Queue desynchronization — same handling as
          // _consume_sub_confirmation.
          _protocol_error_and_close()
        end
        _mode = RedisNormalMode
      end
      // count > 0: more confirmations coming, don't deliver yet.
    else
      // Explicit unsubscribe with known channel count. Use the shared
      // countdown logic.
      _consume_sub_confirmation(arr)

      // Check if subscription count reached zero — only transition
      // back to normal mode if there are no more pending subscribe/
      // unsubscribe calls in the queue.
      let count_value: I64 =
        try
          match arr.values(2)?
          | let count: RedisInteger => count.value
          else return // Non-integer count — ignore (not fatal for explicit)
          end
        else return
        end

      if (count_value == 0)
        and (_pending_sub_confirmations.size() == 0)
      then
        _mode = RedisNormalMode
      end
    end

  // ============================================================
  // Internal: RESP serialization and sending
  // ============================================================

  fun ref _send_command(args: Array[ByteSeq] val):
    (RedisRequestToken | RedisSendError)
  =>
    """
    Serialize args to RESP format and send. One send() call per command
    (not batched) so that SendToken correlation is one-to-one.
    """
    let resp_data = _serialize_command(args)

    match _tcp_connection.send(consume resp_data)
    | let send_token: SendToken =>
      _next_token_id = _next_token_id + 1
      let redis_token = RedisRequestToken._create(_next_token_id)
      _pending_requests.push(redis_token)
      _send_token_map(send_token.id) = redis_token
      redis_token
    | SendErrorNotConnected => RedisSendNotConnected
    | SendErrorNotWriteable => RedisSendNotWriteable
    end

  fun _serialize_command(args: Array[ByteSeq] val): Array[U8] iso^ =>
    """
    Serialize a command to RESP array-of-bulk-strings format.

    Example: ["SET", "key", "value"] becomes:
    *3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n

    Uses _RespSerializer primitive for helper methods because `this`
    becomes `tag` inside a recover block, which can't satisfy the `box`
    receiver needed by instance methods.
    """
    _RespSerializer.serialize_command(args)

// RESP serialization helpers. Extracted into a primitive so they can be
// called from inside recover blocks (where `this` is `tag` and can't
// satisfy the `box` receiver needed by instance methods).
primitive _RespSerializer
  fun serialize_command(args: Array[ByteSeq] val): Array[U8] iso^ =>
    """
    Serialize a command to RESP array-of-bulk-strings format.
    """
    // Calculate total size for single allocation
    var total: USize = 1 + _digits(args.size()) + 2  // *N\r\n
    for arg in args.values() do
      total = total + 1 + _digits(arg.size()) + 2  // $N\r\n
      total = total + arg.size() + 2                // data\r\n
    end

    recover
      let buf = Array[U8](total)
      buf.push('*')
      _append_number(buf, args.size())
      buf.push('\r')
      buf.push('\n')
      for arg in args.values() do
        buf.push('$')
        _append_number(buf, arg.size())
        buf.push('\r')
        buf.push('\n')
        match arg
        | let s: String =>
          buf.append(s.array())
        | let a: Array[U8] val =>
          buf.append(a)
        end
        buf.push('\r')
        buf.push('\n')
      end
      buf
    end

  fun _digits(n: USize): USize =>
    """Number of decimal digits in n."""
    if n == 0 then return 1 end
    var count: USize = 0
    var v = n
    while v > 0 do
      count = count + 1
      v = v / 10
    end
    count

  fun _append_number(buf: Array[U8], n: USize) =>
    """Append the decimal representation of n to buf."""
    let s: String val = n.string()
    buf.append(s.array())

// Mode tracking — public so users can query mode() and match on the result
primitive RedisNormalMode
primitive RedisPubSubMode
type RedisMode is (RedisNormalMode | RedisPubSubMode)

// Internal parse result types
primitive _RespIncomplete
primitive _RespParseError
type _RespParseResult is (_RespIncomplete | _RespParseError)

RedisClientCallbacks — user-implemented trait

trait RedisClientCallbacks
  """
  Protocol-level callbacks for client-side Redis connections. The user's
  actor implements this trait to receive typed Redis events. All callbacks
  run inside the user's actor — direct access to application state, no
  actor boundary overhead.
  """

  fun ref _on_redis_connected() =>
    """
    The connection is ready for Redis commands. For plaintext connections,
    this fires when TCP connects. For TLS connections, this fires after the
    TLS handshake completes.
    """
    None

  fun ref _on_redis_connection_failure() =>
    """
    The connection could not be established, or the TLS handshake failed.
    The application was never notified via `_on_redis_connected`.
    """
    None

  fun ref _on_redis_response(token: RedisRequestToken, value: RedisValue) =>
    """
    A response to a command identified by `token`. The value is the parsed
    RESP response — could be any RESP type including RedisError (which is
    a Redis-level error like "WRONGTYPE", not a protocol layer error).

    In normal mode, responses arrive in the same order commands were sent
    (pipelining). In pub/sub mode, subscription/unsubscription
    confirmations are delivered here.
    """
    None

  fun ref _on_redis_push(channel: String, message: Array[U8] val) =>
    """
    A pub/sub message received on `channel`. Only fires in pub/sub mode
    after a successful `subscribe()`.
    """
    None

  fun ref _on_redis_closed() =>
    """
    The connection has been closed. No further callbacks will fire except
    possibly `_on_redis_send_failed` for any pending partial writes.
    """
    None

  fun ref _on_redis_throttled() =>
    """
    The TCP socket is under backpressure. `command()` will return
    `RedisSendNotWriteable` until `_on_redis_unthrottled` fires. The
    application should stop sending commands and queue or drop them.
    """
    None

  fun ref _on_redis_unthrottled() =>
    """
    Backpressure has been released. `command()` will accept sends again.
    """
    None

  fun ref _on_redis_sent(token: RedisRequestToken) =>
    """
    The bytes for command `token` have been fully handed to the OS. This
    does NOT mean Redis has processed the command — only that the data
    left the application. `_on_redis_response` fires when Redis replies.

    Useful for latency measurement (time between sent and response) or for
    confirming that a fire-and-forget command has at least left the
    application.
    """
    None

  fun ref _on_redis_send_failed(token: RedisRequestToken) =>
    """
    The bytes for command `token` were accepted by `command()` but could
    not be delivered to the OS (connection closed with a partial write
    pending). Always arrives after `_on_redis_closed`.
    """
    None

  fun ref _on_redis_protocol_error(err: RedisProtocolError) =>
    """
    The RESP parser encountered malformed data. The connection will be
    closed immediately after this callback (hard_close). The byte stream
    is desynchronized and cannot be recovered.
    """
    None

RedisClientActor — convenience trait

trait RedisClientActor is TCPConnectionActor
  """
  Convenience trait that provides a default `_connection()` implementation,
  eliminating the `_connection()` boilerplate. The user implements this
  instead of `TCPConnectionActor` directly.

  The override hazard is minimal: `_connection()` and the ASIO behaviors
  (`_event_notify`, `_read_again`, etc.) are Lori internals that users
  would never think to override, unlike lifecycle callbacks like
  `_on_received` which have protocol-level meaning (and which live on the
  interceptor class, not this trait).
  """

  fun ref _redis_client(): RedisClient
  fun ref _connection(): TCPConnection => _redis_client().connection()

RedisServer — server-side protocol class

class RedisServer is ServerLifecycleEventReceiver
  """
  Lifecycle interceptor for server-side Redis connections. Same pattern
  as RedisClient but for the server side: implements
  ServerLifecycleEventReceiver, intercepts raw byte callbacks, parses
  incoming RESP commands, and delivers them to the user via
  RedisServerCallbacks.

  Used when building a Redis-protocol-compatible server (or a service
  that speaks the Redis wire protocol).
  """

  var _tcp_connection: TCPConnection = TCPConnection.none()
  let _parse_buffer: Array[U8] = Array[U8]
  var _parse_offset: USize = 0
  let _callbacks: (RedisServerCallbacks ref | None)

  // Send token tracking (same pattern as RedisClient)
  var _next_token_id: USize = 0
  let _send_token_map: Map[USize, RedisRequestToken] =
    Map[USize, RedisRequestToken]

  new create(auth: TCPServerAuth,
    fd: U32,
    enclosing: TCPConnectionActor ref,
    callbacks: RedisServerCallbacks ref)
  =>
    _callbacks = callbacks
    _tcp_connection = TCPConnection.server(auth, fd, enclosing, this)

  new ssl(auth: TCPServerAuth,
    ssl_ctx: SSLContext val,
    fd: U32,
    enclosing: TCPConnectionActor ref,
    callbacks: RedisServerCallbacks ref)
  =>
    _callbacks = callbacks
    _tcp_connection = TCPConnection.ssl_server(auth, ssl_ctx, fd, enclosing,
      this)

  new none() =>
    _callbacks = None

  fun ref connection(): TCPConnection => _tcp_connection

  fun ref reply(value: RedisValue): (RedisRequestToken | RedisSendError) =>
    """
    Send a RESP value as a response to the client. Used for responding
    to commands received via `_on_redis_command`.
    """
    let resp_data = _serialize_value(value)
    match _tcp_connection.send(consume resp_data)
    | let send_token: SendToken =>
      _next_token_id = _next_token_id + 1
      let redis_token = RedisRequestToken._create(_next_token_id)
      _send_token_map(send_token.id) = redis_token
      redis_token
    | SendErrorNotConnected => RedisSendNotConnected
    | SendErrorNotWriteable => RedisSendNotWriteable
    end

  fun ref push(channel: String, message: Array[U8] val):
    (RedisRequestToken | RedisSendError)
  =>
    """
    Send a pub/sub message to the client. Used when the server is
    maintaining subscriptions and needs to push messages.
    """
    // Serialize as ["message", channel, payload] RESP array
    let args: Array[ByteSeq] val = recover
      ["message"; channel; message]
    end
    let resp_data = _serialize_command(args)
    match _tcp_connection.send(consume resp_data)
    | let send_token: SendToken =>
      _next_token_id = _next_token_id + 1
      let redis_token = RedisRequestToken._create(_next_token_id)
      _send_token_map(send_token.id) = redis_token
      redis_token
    | SendErrorNotConnected => RedisSendNotConnected
    | SendErrorNotWriteable => RedisSendNotWriteable
    end

  // --- ServerLifecycleEventReceiver implementation ---

  fun ref _connection(): TCPConnection => _tcp_connection

  fun ref _on_started() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_started()
    end

  fun ref _on_start_failure() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_start_failure()
    end

  fun ref _on_closed() =>
    _parse_buffer.clear()
    _parse_offset = 0
    // Do NOT clear _send_token_map here — same reasoning as RedisClient:
    // Lori's _on_send_failed fires after _on_closed, and needs the map
    // for SendToken → RedisRequestToken correlation.
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_closed()
    end

  fun ref _on_received(data: Array[U8] iso) =>
    // Parse incoming RESP commands (same parser as RedisClient)
    // Redis commands are RESP arrays of bulk strings
    _feed(consume data)

  fun ref _on_throttled() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_throttled()
    end

  fun ref _on_unthrottled() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_unthrottled()
    end

  fun ref _on_sent(token: SendToken) =>
    try
      let redis_token = _send_token_map(token.id)?
      _send_token_map.remove(token.id)?
      match _callbacks
      | let cb: RedisServerCallbacks ref => cb._on_redis_sent(redis_token)
      end
    end

  fun ref _on_send_failed(token: SendToken) =>
    try
      let redis_token = _send_token_map(token.id)?
      _send_token_map.remove(token.id)?
      match _callbacks
      | let cb: RedisServerCallbacks ref => cb._on_redis_send_failed(redis_token)
      end
    end

  fun ref _on_tls_ready() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_started()
    end

  fun ref _on_tls_failure() =>
    match _callbacks
    | let cb: RedisServerCallbacks ref => cb._on_redis_start_failure()
    end

  // --- Internal: parsing and serialization ---
  // (Same RESP parser as RedisClient, but dispatches parsed commands
  // to _on_redis_command instead of matching responses to requests.)

  fun ref _feed(data: Array[U8] iso) =>
    // Same parse loop as RedisClient._feed, but _dispatch_value calls
    // _callbacks._on_redis_command(value) for each complete command.
    None // placeholder — shares parser implementation with RedisClient

  fun ref _serialize_value(value: RedisValue): Array[U8] iso^ =>
    """
    Serialize any RedisValue to RESP wire format for sending as a
    response.
    """
    recover Array[U8] end // placeholder

  fun ref _serialize_command(args: Array[ByteSeq] val): Array[U8] iso^ =>
    // Same as RedisClient._serialize_command
    recover Array[U8] end // placeholder

RedisServerCallbacks — user-implemented trait

trait RedisServerCallbacks
  """
  Protocol-level callbacks for server-side Redis connections.
  """

  fun ref _on_redis_started() =>
    """
    The server connection is ready. For plaintext connections, this fires
    when the accepted TCP connection is established. For TLS, this fires
    after the handshake completes.
    """
    None

  fun ref _on_redis_start_failure() =>
    """
    The server connection failed to start (e.g., TLS handshake failure).
    """
    None

  fun ref _on_redis_command(value: RedisValue) =>
    """
    A complete Redis command has been received from the client. The value
    is a RESP array of bulk strings (the standard Redis command format).
    The server should process the command and call `reply()` to respond.
    """
    None

  fun ref _on_redis_closed() =>
    """
    The connection has been closed.
    """
    None

  fun ref _on_redis_throttled() =>
    None

  fun ref _on_redis_unthrottled() =>
    None

  fun ref _on_redis_sent(token: RedisRequestToken) =>
    None

  fun ref _on_redis_send_failed(token: RedisRequestToken) =>
    None

  fun ref _on_redis_protocol_error(err: RedisProtocolError) =>
    None

RedisServerActor — convenience trait

trait RedisServerActor is TCPConnectionActor
  fun ref _redis_server(): RedisServer
  fun ref _connection(): TCPConnection => _redis_server().connection()

Usage examples

Client — echo back responses

actor MyRedisClient is (RedisClientActor & RedisClientCallbacks)
  var _redis: RedisClient = RedisClient.none()

  new create(auth: TCPConnectAuth, host: String, port: String) =>
    _redis = RedisClient(auth, host, port, this, this)

  fun ref _redis_client(): RedisClient => _redis

  fun ref _on_redis_connected() =>
    _redis.command(["SET"; "greeting"; "hello world"])

  fun ref _on_redis_response(token: RedisRequestToken, value: RedisValue) =>
    match value
    | let s: RedisSimpleString =>
      // "OK" from SET
      _redis.command(["GET"; "greeting"])
    | let b: RedisBulkString =>
      // "hello world" from GET
      // do something with b.value
      _redis.connection().close()
    | let e: RedisError =>
      // Redis error response
      // handle e.message
      None
    end

Client with TLS

actor MyTLSRedisClient is (RedisClientActor & RedisClientCallbacks)
  var _redis: RedisClient = RedisClient.none()

  new create(auth: TCPConnectAuth, ssl_ctx: SSLContext val,
    host: String, port: String)
  =>
    _redis = RedisClient.ssl(auth, ssl_ctx, host, port, this, this)

  fun ref _redis_client(): RedisClient => _redis

  fun ref _on_redis_connected() =>
    // TLS handshake is complete, connection is encrypted
    _redis.command(["AUTH"; "password"])

  fun ref _on_redis_connection_failure() =>
    // TLS handshake failed or TCP connection failed
    None

Client with pub/sub

actor MySubscriber is (RedisClientActor & RedisClientCallbacks)
  var _redis: RedisClient = RedisClient.none()
  let _out: OutStream

  new create(auth: TCPConnectAuth, host: String, port: String,
    out: OutStream)
  =>
    _out = out
    _redis = RedisClient(auth, host, port, this, this)

  fun ref _redis_client(): RedisClient => _redis

  fun ref _on_redis_connected() =>
    _redis.subscribe(["events"; "notifications"])

  fun ref _on_redis_push(channel: String, message: Array[U8] val) =>
    _out.print("Channel " + channel + ": " + String.from_array(message))

  fun ref _on_redis_response(token: RedisRequestToken, value: RedisValue) =>
    // Subscription confirmations arrive here
    None

Server — echo server speaking Redis protocol

actor Listener is TCPListenerActor
  var _tcp_listener: TCPListener = TCPListener.none()
  let _server_auth: TCPServerAuth

  new create(listen_auth: TCPListenAuth) =>
    _server_auth = TCPServerAuth(listen_auth)
    _tcp_listener = TCPListener(listen_auth, "127.0.0.1", "6379", this)

  fun ref _listener(): TCPListener => _tcp_listener

  fun ref _on_accept(fd: U32): RedisEchoHandler =>
    RedisEchoHandler(_server_auth, fd)

  fun ref _on_listening() => None
  fun ref _on_listen_failure() => None

actor RedisEchoHandler is (RedisServerActor & RedisServerCallbacks)
  var _redis: RedisServer = RedisServer.none()

  new create(auth: TCPServerAuth, fd: U32) =>
    _redis = RedisServer(auth, fd, this, this)

  fun ref _redis_server(): RedisServer => _redis

  fun ref _on_redis_command(value: RedisValue) =>
    // Echo the command back as a response
    _redis.reply(value)

Design decisions and rationale

Callbacks as trait vs interface

RedisClientCallbacks and RedisServerCallbacks are traits (nominal subtyping). The user declares is RedisClientCallbacks. We chose traits over interfaces because:

• The relationship is intentional and discoverable — is RedisClientCallbacks in the actor declaration tells readers this actor handles Redis callbacks.
• The callback methods have _on_redis_ prefixes (private by convention), which only makes sense for a deliberately adopted trait, not an accidentally satisfied interface.
• If a user wants to handle both Redis and HTTP on one actor, they declare is (RedisClientCallbacks & HttpClientCallbacks) — explicit about what they're doing.

_on_connecting is absorbed

ClientLifecycleEventReceiver._on_connecting(inflight_connections) reports Happy Eyeballs connection progress. This is TCP-level detail irrelevant to Redis protocol operation. The interceptor absorbs it silently. If a user wants this information, they can access the TCPConnection directly via _redis.connection() — but this is an escape hatch, not the normal path.

_on_sent/_on_send_failed are correlated and forwarded

The interceptor receives Lori's SendToken callbacks and translates them to RedisRequestToken callbacks. This hides SendToken from the user while preserving the semantic: "your command bytes reached the OS" (_on_redis_sent) and "your command bytes couldn't be delivered" (_on_redis_send_failed). The one-to-one command-to-send mapping (one send() per command) makes correlation straightforward — each SendToken maps to exactly one RedisRequestToken.

One send() per command (not batched)

Each command() call serializes and sends immediately via a single _tcp_connection.send(). This gives clean one-to-one SendToken-to-RedisRequestToken correlation. The cost is more syscalls for pipelined commands. The benefit is that _on_redis_sent and _on_redis_send_failed are precise — they refer to exactly one command, not a batch.

Batching (multiple pipelined commands in one send()) could be added as an optimization, but it complicates _on_send_failed — if a batch fails, all commands in the batch failed, and the per-command tokens can't distinguish which. Starting with one-to-one is simpler and correct; batching can be added later if profiling shows it matters.

Pub/sub mode transition is guarded

subscribe() returns RedisSendInvalidMode if there are pending pipelined responses in normal mode. This prevents desynchronization: if you have 3 pending normal-mode responses and enter pub/sub mode, the parser doesn't know whether the next RESP value is a normal response or a subscription confirmation. Requiring the response queue to be empty before transitioning ensures clean mode boundaries.

unsubscribe() returns RedisSendInvalidMode if called in normal mode. The back-transition from pub/sub to normal happens automatically when the subscription count reaches zero (detected from the unsubscribe confirmation's count field).

Multi-channel subscription confirmation tracking

A single subscribe(["a", "b"]) call sends one SUBSCRIBE command with two channels but Redis replies with two separate confirmations — one per channel. The protocol class tracks this with a queue (_pending_sub_confirmations), where each entry is the number of confirmations expected for one subscribe/unsubscribe call. As confirmations arrive, the front entry is decremented. When it reaches zero, the matching RedisRequestToken is dequeued from _pending_requests and the user receives _on_redis_response with the final confirmation.

This queue-based approach (rather than a single counter) correctly handles interleaved calls — e.g., subscribe(["a", "b"]) followed by subscribe(["c"]) before any confirmations arrive pushes [2, 1] onto the queue, and the two sets of confirmations are correctly associated with their respective tokens.

Unsubscribe-all: unsubscribe([]) (empty array = unsubscribe from all channels) pushes 0 as a sentinel. _consume_sub_confirmation returns early when it sees the sentinel, deferring to _dispatch_unsubscription_confirmation which inspects the subscription count in each confirmation. When the count reaches zero, the token is delivered via _on_redis_response and the connection transitions back to normal mode. Intermediate confirmations (count > 0) are consumed without delivery — the user gets a single _on_redis_response when unsubscribe-all is complete, consistent with explicit unsubscribe behavior.

Unexpected values in pub/sub mode

In pub/sub mode, only RESP arrays are expected (push messages and sub/unsub confirmations). Any non-array value — including error responses (-ERR ...), simple strings, integers, etc. — is treated as a protocol error, triggering _on_redis_protocol_error followed by hard_close(). This is conservative: Redis error responses in pub/sub mode indicate something has gone wrong (e.g., an invalid command was sent while in pub/sub mode), and there's no request token to correlate them with. A less aggressive alternative would be a separate _on_redis_pub_sub_error callback, but the use case is narrow enough to defer.

Parse buffer is self-managed (not using expect())

The RESP parser maintains its own buffer and parse offset. It does not use Lori's expect() because:

• expect() can only express "give me exactly N bytes" — it can't express "read until \r\n," which is needed for RESP type prefixes.
• RESP's variable-length framing (type prefixes are terminated by \r\n, not length-prefixed) means expect() could only help with bulk string bodies, not with the overall parse flow.
• Self-managed buffering gives the protocol layer complete control over parse state, making it independent of Lori's read machinery.

Protocol errors trigger hard close

When the RESP parser encounters malformed data, it delivers _on_redis_protocol_error followed by hard_close(). No recovery is attempted — a parse error means the byte stream is desynchronized, and subsequent bytes would be interpreted at wrong boundaries. This follows Lori's own pattern (SSL errors trigger hard_close with failure callbacks).

AUTH is not built into the protocol layer

AUTH is a normal Redis command. The user sends it manually via command(["AUTH"; "password"]) after _on_redis_connected. Building AUTH into the constructor would add complexity (deferred connected callback, error handling for AUTH failure) for a feature that's one line of user code. If automatic AUTH becomes a common need, it can be added as an optional constructor parameter later without breaking the existing API.

none() and embed considerations

RedisClient.none() exists as a placeholder for field initialization. However, since _finish_initialization (the behavior that accesses _connection()) runs asynchronously after the constructor, _redis is always initialized by the time it's accessed. This means none() may be unnecessary and embed may be viable:

// If none() is unnecessary:
actor MyRedisClient is (RedisClientActor & RedisClientCallbacks)
  embed _redis: RedisClient

  new create(auth: TCPConnectAuth, host: String, port: String) =>
    _redis = RedisClient(auth, host, port, this, this)

embed eliminates heap indirection for the protocol class. The none() constructor (and its use of None for the callbacks field) would also become unnecessary. This should be verified empirically — if Pony's definite assignment analysis accepts the field without none(), embed is the better choice.

Shared RESP parser

RedisClient and RedisServer both need a RESP parser. The implementation should be extracted into a shared RespParser class that both use internally. The parser is a pure state machine: feed bytes in, get RedisValues out. No connection or callback dependencies. This is essentially Shape 5 (parser library) serving as the foundation, with Shape 2 (interceptor) layered on top.

Things this design does NOT handle

• Reconnection: This is a single-connection design. Automatic reconnection, re-subscription to pub/sub channels, and command replay on reconnect are left to a layer above. This keeps the protocol class simple and matches Lori's own single-connection model.
• Connection pooling: Multiple RedisClient instances in separate actors. Pooling logic is above this layer.
• Redis Cluster: Cluster-aware routing, MOVED/ASK redirection, slot mapping. These would build on top of RedisClient.
• RESP3: This design targets RESP2 (the default protocol up to Redis 6). RESP3 adds new types (maps, sets, pushes) that would require extending RedisValue and the parser.
• Pipelining batching: Commands are sent one-at-a-time. A future optimization could buffer commands and flush as a batch, but this complicates SendToken correlation.
• Receive-side flow control: mute()/unmute() on the TCPConnection are available via _redis.connection(), but the protocol layer doesn't provide its own pause/resume for parsed-but-undelivered frames. If needed, this could be added by having the parser buffer complete frames and deliver them on demand.
• In-flight command notification on close: When the connection closes, commands that were successfully sent (_on_redis_sent already fired) but whose responses hadn't arrived are silently discarded. The user gets _on_redis_closed but no per-command notification of which responses were lost. Applications that need this (e.g., to decide whether to retry a write) must track in-flight tokens themselves — record tokens at _on_redis_sent, remove them at _on_redis_response. A future _on_redis_response_lost(token) callback could be added if this becomes a common need.