rustzmq2 0.1.0

//!
//! Replaces the earlier two-task `(reader_loop, writer_loop)` pattern
//! with a single `peer_loop` task that owns both the read half and the
//! `VectoredWriter`. This halves per-peer scheduling overhead at scale.

use crate::async_rt::notify::AsyncNotify;
use crate::async_rt::time::{DefaultClock, RuntimeClock};
use crate::codec::{CodecError, Message};
use crate::engine::registry::PeerKey;
use crate::engine::writer::VectoredWriter;
use crate::engine::{FlushState, Outbound, TaggedInboundTx};
use crate::io_compat::AsyncVectoredWrite;
#[cfg(feature = "curve")]
use crate::mechanism::{build_nonce, CurveSession};
#[cfg(feature = "curve")]
use crate::message::ZmqMessage;

#[cfg(feature = "curve")]
use bytes::Bytes;
#[cfg(feature = "curve")]
use crypto_box::aead::Aead;
use futures::channel::oneshot;
use futures::future::Shared;
use futures::{FutureExt, Stream, StreamExt};
use rand::RngExt;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use crate::engine::HeartbeatConfig;

pub(crate) type PeerWriterKind<W> = VectoredWriter<W>;

/// Per-peer last-message slot for the CONFLATE receive path.
/// The reader loop overwrites this slot on every inbound message;
/// `recv_next` polls all slots and parks on the notify when all are empty.
pub(crate) struct ConflateSlotInner {
    pub slot: parking_lot::Mutex<Option<(PeerKey, crate::message::ZmqMessage)>>,
    /// Shared across all peers on the same socket; woken on every slot write.
    pub notify: Arc<crate::async_rt::notify::RuntimeNotify>,
}

pub(crate) type ConflateSlot = Arc<ConflateSlotInner>;

/// Optional per-peer configuration passed into the peer loop.
pub(crate) struct PeerConfig {
    pub heartbeat: Option<HeartbeatConfig>,
    #[cfg(feature = "curve")]
    pub curve: Option<CurveSession>,
    pub max_msg_size: Option<usize>,
    /// When Some, the peer loop writes into the slot instead of the shared inbound channel.
    pub conflate_slot: Option<ConflateSlot>,
    /// Max payload bytes coalesced per writer `writev(2)`
    /// (`ZMQ_OUT_BATCH_SIZE`). `None` disables the byte budget — only
    /// the message-count ceiling (`out_batch_msgs`) applies. Byte-based
    /// to match libzmq's semantics — see `VectoredWriter::drain_batch`.
    pub out_batch_size: Option<usize>,
    /// Hard ceiling on messages drained per batch. `None` disables the
    /// message-count ceiling — only `out_batch_size` applies. Caps
    /// per-peer pending-queue depth so tiny-message high-fanout workloads
    /// can't spiral into HWM overflow on PUB.
    pub out_batch_msgs: Option<usize>,
    /// Max messages the reader drains from the framed decoder per
    /// scheduler wake. `None` disables batching (one message per wake).
    /// Unlike libzmq's byte-based `ZMQ_IN_BATCH_SIZE`, this is a message
    /// count; the codec already byte-buffers underneath.
    pub in_batch_msgs: Option<usize>,
    /// Caller-thread inline-write fast path enable + payload cap.
    /// Outer `None` disables inline (channel-only). Outer `Some(inner)`
    /// enables inline; the inner `Option<usize>` is the per-message
    /// payload cap (`None` = no cap, any size accepted; `Some(n)` =
    /// decline payloads `>= n`). CURVE engines force this to `None`
    /// — the inline path bypasses the cipher state in `peer_loop`.
    #[allow(clippy::option_option)]
    pub inline_write_max: Option<Option<usize>>,
}

impl Default for PeerConfig {
    fn default() -> Self {
        Self {
            heartbeat: None,
            #[cfg(feature = "curve")]
            curve: None,
            max_msg_size: None,
            conflate_slot: None,
            out_batch_size: Some(8192),
            out_batch_msgs: Some(32),
            // Reader batching off by default — see SocketOptions for
            // rationale. Users can opt in via the builder.
            in_batch_msgs: None,
            // Inline-write off by default at the engine layer; the
            // socket layer (SocketCore::new) resolves the per-type
            // default and overrides this for REQ/REP/PAIR.
            inline_write_max: None,
        }
    }
}

/// Channel endpoints that bridge the peer loop to the socket layer.
pub(crate) struct PeerChannels {
    pub outbound_rx: flume::Receiver<Outbound>,
    pub shared_inbound: TaggedInboundTx,
}

/// Single unified task per peer.
///
/// Generic over `R` (the read half, a Stream) and `W` (the vectored writer
/// write half). This lets both tokio and smol transports share the same loop
/// body — only the concrete `R`/`W` types differ at the call site.
pub(crate) async fn peer_loop<R, W>(
    read_half: R,
    writer: PeerWriterKind<W>,
    channels: PeerChannels,
    peer_key: PeerKey,
    flush_state: Arc<FlushState>,
    shutdown: Shared<oneshot::Receiver<()>>,
    config: PeerConfig,
) where
    R: Stream<Item = Result<Message, CodecError>> + Unpin + Send + 'static,
    W: AsyncVectoredWrite + Send + 'static,
{
    let result = peer_loop_inner(
        read_half,
        writer,
        channels,
        peer_key,
        &flush_state,
        shutdown,
        config,
    )
    .await;
    let _ = result;
    // Always mark writer dead so flush waiters don't hang. Use the
    // gated notify; the writer_alive=false store happens before the
    // flush_waiters read inside notify_flush_waiters, so no waiter
    // can register-then-park without observing the dead flag.
    flush_state.writer_alive.store(false, Ordering::Release);
    flush_state.notify_flush_waiters();
}

async fn peer_loop_inner<R, W>(
    mut read_half: R,
    mut writer: PeerWriterKind<W>,
    channels: PeerChannels,
    peer_key: PeerKey,
    flush_state: &Arc<FlushState>,
    shutdown: Shared<oneshot::Receiver<()>>,
    config: PeerConfig,
) -> std::io::Result<()>
where
    R: Stream<Item = Result<Message, CodecError>> + Unpin,
    W: AsyncVectoredWrite + Send + Sync + 'static,
{
    // ── Heartbeat state ────────────────────────────────────────────────────
    let PeerChannels {
        outbound_rx,
        shared_inbound,
    } = channels;
    let PeerConfig {
        heartbeat: heartbeat_cfg,
        #[cfg(feature = "curve")]
        curve,
        max_msg_size,
        conflate_slot,
        out_batch_size,
        out_batch_msgs,
        in_batch_msgs,
        inline_write_max,
    } = config;
    // Engine opted into caller-thread inline writes; gate the
    // overflow-poll arm and busy-flag bookkeeping on this.
    let inline_enabled = inline_write_max.is_some();
    let mut hb: Option<HeartbeatState<DefaultClock>> = heartbeat_cfg.map(HeartbeatState::new);
    #[cfg(feature = "curve")]
    let mut curve = curve;
    // Fuse shutdown once; futures::select! requires FusedFuture.
    let mut shutdown = shutdown.fuse();
    // Accumulate decrypted frames until the last MESSAGE (more=0) arrives.
    #[cfg(feature = "curve")]
    let mut curve_recv_buf: Vec<Bytes> = Vec::new();

    loop {
        crate::wake_counter::bump(&crate::wake_counter::PEER_LOOP_ITERS);
        // Adopt any inline-path partial-write remainders into the front
        // of our pending queue. Wire-order rule: an interrupted message
        // must resume before any new write goes out. Cheap (atomic load
        // + branch) when overflow is empty (the steady state). Skipped
        // entirely when inline is disabled — no caller can produce
        // overflow.
        if inline_enabled {
            writer.pull_inline_overflow();
        }
        // ── Write drain ────────────────────────────────────────────────────
        // Try one non-blocking flush pass, then select concurrently: waiting
        // for socket writable MUST NOT block reads/heartbeats/shutdown on
        // this task. Blocking here deadlocks DEALER↔ROUTER over IPC when
        // both sides fill their kernel send buffers — neither can drain
        // the other's write queue because neither is polling the reader.
        if !writer.is_empty() {
            let flushed = writer.flush_one_pass()?;
            if flushed > 0 {
                flush_state
                    .flushed
                    .fetch_add(flushed as u64, Ordering::Release);
                flush_state.notify_flush_waiters();
            }
        }

        // When the writer has pending bytes, park on `writable_fut`
        // and suppress the outbound arm — that preserves HWM
        // backpressure (sender's `send().await` blocks on the
        // bounded outbound channel) while reads still fire on the
        // read arm.
        //
        // Each arm is `Either<RealFut, Pending>` so the disabled case
        // is a never-resolving placeholder; this stack-pins both
        // halves and avoids per-iter `Box::pin` heap allocations.
        let pending_drain = !writer.is_empty();
        use futures::future::Either;
        let writable_fut = if pending_drain {
            Either::Left(writer.writable_owned())
        } else {
            Either::Right(std::future::pending::<std::io::Result<()>>())
        };
        let outbound_fut = if !pending_drain {
            Either::Left(outbound_rx.recv_async())
        } else {
            Either::Right(std::future::pending::<
                Result<crate::engine::Outbound, flume::RecvError>,
            >())
        };
        let hb_sleep_fut = match hb {
            Some(ref mut h) => Either::Left(h.next_ping_sleep()),
            None => Either::Right(std::future::pending::<()>()),
        };
        // Inline-path overflow notify — only armed on engines that
        // opted into inline writes; disabled engines can't generate
        // overflow so the arm is `Pending`.
        let overflow_fut = if inline_enabled {
            Either::Left(writer.overflow_notified())
        } else {
            Either::Right(std::future::pending::<()>())
        };
        futures::pin_mut!(writable_fut, outbound_fut, hb_sleep_fut, overflow_fut);

        // ── Select: read | new outbound | writable | heartbeat | shutdown ──
        futures::select! {
            ready = writable_fut.fuse() => {
                crate::wake_counter::bump(&crate::wake_counter::PEER_LOOP_WRITABLE_WAKES);
                // Either::Right(pending) never resolves, so this only fires
                // when pending_drain was true (Either::Left).
                ready?;
                // Socket drained enough — fall through to next loop iteration
                // which will re-run flush_one_pass at the top.
            }
            msg = read_half.next().fuse() => {
                crate::wake_counter::bump(&crate::wake_counter::PEER_LOOP_READ_WAKES);
                // Process the first message that woke us, then try to drain
                // up to `in_batch_msgs - 1` additional synchronously-ready
                // messages from the codec without returning to the outer
                // select. Mirrors libzmq's `stream_engine_base_t::in_event`
                // inner decode loop — one socket read fans out into N
                // message deliveries before we yield to other arms.
                //
                // Batching stops at the first `Poll::Pending` (no more
                // decoded data buffered), EOF, or budget exhaustion.
                // Heartbeat/shutdown arms remain responsive because the
                // budget caps how long we stay in this arm.
                let budget = in_batch_msgs.unwrap_or(1).max(1);
                let mut msg = msg;
                let mut drained: usize = 0;
                loop {
                    let should_continue = process_one_read_message(
                        msg,
                        &mut hb,
                        #[cfg(feature = "curve")]
                        &mut curve,
                        #[cfg(feature = "curve")]
                        &mut curve_recv_buf,
                        max_msg_size,
                        &shared_inbound,
                        conflate_slot.as_deref(),
                        peer_key,
                        &mut writer,
                    ).await?;
                    if !should_continue {
                        return Ok(());
                    }
                    drained += 1;
                    if drained >= budget {
                        break;
                    }
                    // Peek the next message synchronously. `now_or_never`
                    // polls the future once with the current waker; if
                    // it returns `Poll::Pending` we get `None` back and
                    // break out to the outer select so other arms fire.
                    match read_half.next().now_or_never() {
                        Some(next_msg) => msg = next_msg,
                        None => break,
                    }
                }
            }
            item = outbound_fut.fuse() => {
                crate::wake_counter::bump(&crate::wake_counter::PEER_LOOP_OUTBOUND_WAKES);
                match item {
                    Ok(o) => {
                        // Hold the busy flag from dequeue to
                        // disposition so a racing caller-thread inline
                        // write can't jump FIFO. No-op when inline is
                        // disabled (no racing caller exists).
                        if inline_enabled {
                            writer.mark_peer_loop_busy();
                        }
                        // This arm only fires when `pending_drain == false`,
                        // so `writer.is_empty()` holds and the fast path is
                        // safe to attempt.
                        let msg = o.msg;
                        #[cfg(feature = "curve")]
                        {
                            if let Some(sess) = curve.as_mut() {
                                if let Message::Message(zm) = msg {
                                    let n = zm.len();
                                    for (i, frame) in zm.iter().enumerate() {
                                        let more = i < n - 1;
                                        let wire = curve_encrypt_frame(sess, frame, more)?;
                                        writer.enqueue(Message::SecurityRaw(wire));
                                    }
                                } else {
                                    writer.enqueue(msg);
                                }
                            } else {
                                use crate::engine::writer::FastPath;
                                match writer.try_fast_path_single_frame(msg)? {
                                    FastPath::Sent => {
                                        flush_state.flushed.fetch_add(1, Ordering::Release);
                                        flush_state.notify_flush_waiters();
                                    }
                                    FastPath::Enqueued => {}
                                    FastPath::NotTaken(msg) => {
                                        writer.enqueue(msg);
                                        writer.drain_batch(&outbound_rx, out_batch_size, out_batch_msgs);
                                    }
                                }
                            }
                        }
                        #[cfg(not(feature = "curve"))]
                        {
                            use crate::engine::writer::FastPath;
                            match writer.try_fast_path_single_frame(msg)? {
                                FastPath::Sent => {
                                    flush_state.flushed.fetch_add(1, Ordering::Release);
                                    flush_state.notify_flush_waiters();
                                }
                                FastPath::Enqueued => {}
                                FastPath::NotTaken(msg) => {
                                    writer.enqueue(msg);
                                    writer.drain_batch(&outbound_rx, out_batch_size, out_batch_msgs);
                                }
                            }
                        }
                        if inline_enabled {
                            writer.clear_peer_loop_busy();
                        }
                    }
                    Err(_) => return Ok(()),
                }
            }
            _ = hb_sleep_fut.fuse() => {
                if let Some(ref mut h) = hb {
                    match h.on_ping_tick(&mut writer) {
                        HeartbeatAction::Evict => return Ok(()),
                        HeartbeatAction::Continue => {}
                    }
                }
            }
            _ = overflow_fut.fuse() => {
                // Partial-write recovery wakeup. The actual adoption
                // happens in `pull_inline_overflow()` at the top of
                // the next iteration.
            }
            _ = shutdown => return Ok(()),
        }
    }
}

// ── Heartbeat state machine ────────────────────────────────────────────────────

use crate::codec::HeartbeatFrame;

enum HeartbeatAction {
    Continue,
    Evict,
}

/// Tracks interval, pending-PONG state, and timeout deadline per peer.
/// Generic over `C: RuntimeClock` so the time source is swappable per runtime.
struct HeartbeatState<C: RuntimeClock> {
    cfg: HeartbeatConfig,
    next_ping_at: C::Instant,
    pong_deadline: Option<C::Instant>,
    ttl_tenths: u16,
}

impl<C: RuntimeClock> HeartbeatState<C> {
    fn new(cfg: HeartbeatConfig) -> Self {
        let ttl_tenths = (cfg.ttl.as_millis() / 100) as u16;
        let next_ping_at = C::now() + cfg.interval;
        Self {
            cfg,
            next_ping_at,
            pong_deadline: None,
            ttl_tenths,
        }
    }

    fn next_ping_sleep(
        &mut self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
        let now = C::now();
        let deadline = if let Some(pong_dl) = self.pong_deadline {
            pong_dl.min(self.next_ping_at)
        } else {
            self.next_ping_at
        };
        C::sleep_until(deadline.max(now))
    }

    fn on_ping_tick<W: AsyncVectoredWrite>(
        &mut self,
        writer: &mut VectoredWriter<W>,
    ) -> HeartbeatAction {
        let now = C::now();

        if let Some(pong_dl) = self.pong_deadline {
            if now >= pong_dl {
                return HeartbeatAction::Evict;
            }
        }

        if now >= self.next_ping_at {
            let ctx_bytes: [u8; 8] = rand::rng().random();
            let context = bytes::Bytes::copy_from_slice(&ctx_bytes);
            writer.enqueue(Message::Heartbeat(HeartbeatFrame::Ping {
                ttl_tenths: self.ttl_tenths,
                context,
            }));
            self.pong_deadline = Some(now + self.cfg.timeout);
            self.next_ping_at = now + self.cfg.interval;
        }

        HeartbeatAction::Continue
    }

    fn on_pong(&mut self) {
        self.pong_deadline = None;
    }
}

// ── RFC 26 CURVE MESSAGE encryption helpers ───────────────────────────────────

// libzmq MESSAGE wire format:
// - ZMTP DATA frame (flag 0x00/0x01, NOT command 0x04)
// - Body: \x07MESSAGE(8) + nonce_short(8 BE) + ciphertext
// - Plaintext inside box: [flags(1)] + payload
//   flags & 0x01 = more (inner multipart signal)
// - Nonce check: monotonic (received > last_seen), matching libzmq's _cn_peer_nonce scheme

#[cfg(feature = "curve")]
fn curve_encrypt_frame(
    sess: &mut CurveSession,
    payload: &Bytes,
    more: bool,
) -> std::io::Result<Bytes> {
    use bytes::BufMut;
    use crypto_box::aead::Aead;

    let prefix: &[u8; 16] = if sess.is_server {
        b"CurveZMQMESSAGES"
    } else {
        b"CurveZMQMESSAGEC"
    };
    let nonce_ctr = sess.tx_nonce;
    sess.tx_nonce += 1;
    let nonce = build_nonce(prefix, nonce_ctr);

    // Build plaintext: flags(1) + payload — single allocation sized exactly.
    let mut plain = bytes::BytesMut::with_capacity(1 + payload.len());
    plain.put_u8(if more { 0x01u8 } else { 0x00u8 });
    plain.extend_from_slice(payload);

    let ciphertext = sess
        .session_box
        .encrypt(&nonce, plain.as_ref())
        .map_err(|_e| std::io::Error::from(CodecError::CurveEncryptFailed))?;

    // Wire: [flag_byte(s)] [len] \x07MESSAGE [nonce_short(8)] [ciphertext]
    // DATA frame (not command): flag 0x01 if more, 0x00 if last.
    let body = 8 + 8 + ciphertext.len(); // \x07MESSAGE(8) + nonce(8) + cipher
    let header_len = if body > 255 { 10 } else { 2 }; // flag(1) + len(1 or 8+1)
    let mut wire = bytes::BytesMut::with_capacity(header_len + body);
    if body > 255 {
        wire.put_u8(if more { 0x03u8 } else { 0x02u8 }); // long data frame
        wire.put_u64(body as u64);
    } else {
        wire.put_u8(if more { 0x01u8 } else { 0x00u8 }); // short data frame
        wire.put_u8(body as u8);
    }
    wire.put_u8(7u8); // name_len = len("MESSAGE")
    wire.extend_from_slice(b"MESSAGE");
    wire.put_u64(nonce_ctr);
    wire.extend_from_slice(&ciphertext);

    Ok(wire.freeze())
}

#[cfg(feature = "curve")]
fn curve_decrypt_message_frame(
    sess: &mut CurveSession,
    frame: Bytes,
) -> std::io::Result<(Bytes, bool)> {
    // frame = one ZmqMessage frame: \x07MESSAGE(8) + nonce(8) + ciphertext
    // minimum: 8 + 8 + mac(16) + flags(1) = 33
    const MIN_LEN: usize = 33;
    if frame.len() < MIN_LEN {
        return Err(std::io::Error::from(CodecError::MessageFrameTooShort));
    }
    if &frame[..8] != b"\x07MESSAGE" {
        return Err(std::io::Error::from(CodecError::Decode(
            "not a MESSAGE frame",
        )));
    }
    let nonce_ctr = u64::from_be_bytes(frame[8..16].try_into().unwrap());
    // Monotonic nonce check: must be strictly greater than last seen (libzmq: nonce <= _cn_peer_nonce → reject)
    if nonce_ctr <= sess.rx_nonce {
        return Err(std::io::Error::from(CodecError::CurveNonceOutOfOrder));
    }
    let prefix: &[u8; 16] = if sess.is_server {
        b"CurveZMQMESSAGEC"
    } else {
        b"CurveZMQMESSAGES"
    };
    let nonce = build_nonce(prefix, nonce_ctr);
    sess.rx_nonce = nonce_ctr;

    let plain = sess
        .session_box
        .decrypt(&nonce, &frame[16..])
        .map_err(|_e| std::io::Error::from(CodecError::CurveDecryptFailed))?;

    if plain.is_empty() {
        return Err(std::io::Error::from(CodecError::CurveEmptyPlaintext));
    }
    let more = plain[0] & 0x01 != 0;
    // Convert Vec<u8> → Bytes without copying: Bytes::from(Vec) is zero-copy.
    let mut plain = Bytes::from(plain);
    let _ = plain.split_to(1); // drop the flags byte
    Ok((plain, more))
}

/// Run the full read-arm per-message pipeline on one decoded message:
/// heartbeat-pong tracking, optional CURVE decrypt, `MAXMSGSIZE` check,
/// and forwarding to the socket layer via `handle_read`. Returns
/// `Ok(true)` to continue, `Ok(false)` to exit the peer loop (graceful
/// EOF / shared-inbound closed).
///
/// Extracted from the peer-loop read arm so the batched drain can call
/// it repeatedly without duplicating the pipeline. Single-message
/// semantics are identical to the pre-batching code path.
#[allow(clippy::too_many_arguments)]
async fn process_one_read_message<W: AsyncVectoredWrite>(
    msg: Option<Result<Message, CodecError>>,
    hb: &mut Option<HeartbeatState<DefaultClock>>,
    #[cfg(feature = "curve")] curve: &mut Option<CurveSession>,
    #[cfg(feature = "curve")] curve_recv_buf: &mut Vec<bytes::Bytes>,
    max_msg_size: Option<usize>,
    shared_inbound: &TaggedInboundTx,
    conflate_slot: Option<&ConflateSlotInner>,
    peer_key: PeerKey,
    writer: &mut PeerWriterKind<W>,
) -> std::io::Result<bool> {
    // Pong detection: any inbound frame can be a PONG when we're mid-heartbeat.
    {
        let pong_received = matches!(
            msg,
            Some(Ok(Message::Heartbeat(
                crate::codec::HeartbeatFrame::Pong { .. }
            )))
        );
        if pong_received {
            if let Some(ref mut h) = hb {
                h.on_pong();
            }
        }
    }

    // CURVE: libzmq sends encrypted app messages as ZMTP DATA frames
    // (not command flag 0x04), body starts with `\x07MESSAGE` + nonce(8)
    // + ciphertext. The codec delivers each as a `Message::Message`
    // with outer more=0; multipart is signaled by the inner flags byte
    // in the decrypted plaintext[0]. Buffer frames until an envelope
    // completes, then emit the reassembled message.
    #[cfg(feature = "curve")]
    let msg = match (curve, msg) {
        (Some(sess), Some(Ok(Message::Message(ref zm))))
            if zm
                .get(0)
                .is_some_and(|f| f.len() >= 8 && &f[..8] == b"\x07MESSAGE") =>
        {
            let mut emit = None;
            'frames: for frame in zm.iter() {
                if let Ok((payload, more)) = curve_decrypt_message_frame(sess, frame.clone()) {
                    curve_recv_buf.push(payload);
                    if !more {
                        let frames = std::mem::take(curve_recv_buf);
                        let mut out = ZmqMessage::from(frames[0].clone());
                        for f in &frames[1..] {
                            out.push_back(f.clone());
                        }
                        emit = Some(Ok(Message::Message(out)));
                        break 'frames;
                    }
                } else {
                    curve_recv_buf.clear();
                    emit = Some(Err(CodecError::Decode("CURVE decrypt error")));
                    break 'frames;
                }
            }
            match emit {
                Some(result) => Some(result),
                // `more=true` mid-envelope: no whole message yet, just
                // continue the drain (subsequent frames will complete it).
                None => return Ok(true),
            }
        }
        (_, other) => other,
    };

    // Enforce MAXMSGSIZE before forwarding.
    let msg = if let (Some(max), Some(Ok(Message::Message(ref zm)))) = (max_msg_size, &msg) {
        let total: usize = zm.iter().map(|f| f.len()).sum();
        if total > max {
            Some(Err(CodecError::Decode("message exceeds MAXMSGSIZE")))
        } else {
            msg
        }
    } else {
        msg
    };

    handle_read(msg, shared_inbound, conflate_slot, peer_key, writer).await
}

/// Process one decoded frame from the read half.
async fn handle_read<W: AsyncVectoredWrite>(
    msg: Option<Result<Message, CodecError>>,
    shared_inbound: &TaggedInboundTx,
    conflate_slot: Option<&ConflateSlotInner>,
    peer_key: PeerKey,
    writer: &mut PeerWriterKind<W>,
) -> std::io::Result<bool> {
    match msg {
        Some(Ok(frame)) => {
            {
                use crate::codec::HeartbeatFrame;
                if let Message::Heartbeat(ref hb) = frame {
                    if let HeartbeatFrame::Ping { context, .. } = hb {
                        writer.enqueue(Message::Heartbeat(HeartbeatFrame::Pong {
                            context: context.clone(),
                        }));
                    }
                    return Ok(true);
                }
            }
            if let Some(slot) = conflate_slot {
                if let Message::Message(zm) = frame {
                    *slot.slot.lock() = Some((peer_key, zm));
                    slot.notify.notify_one();
                }
            } else if shared_inbound
                .send_async((peer_key, Ok(frame)))
                .await
                .is_err()
            {
                return Ok(false);
            }
        }
        Some(Err(e)) => {
            if conflate_slot.is_some() {
                return Ok(false);
            }
            let _ = shared_inbound.send_async((peer_key, Err(e))).await;
            return Ok(false);
        }
        None => {
            if conflate_slot.is_some() {
                return Ok(false);
            }
            let _ = shared_inbound
                .send_async((peer_key, Err(CodecError::PeerDisconnected)))
                .await;
            return Ok(false);
        }
    }
    Ok(true)
}