// Copyright 2015-2024 Swim Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use bytes::{Buf, BufMut, Bytes, BytesMut};
use std::convert::TryFrom;
use std::fmt::{Debug, Write};
use std::str::Utf8Error;
use swimos_api::address::RelativeAddress;
use swimos_form::read::ReadError;
use swimos_form::read::Recognizer;
use swimos_form::write::StructuralWritable;
use swimos_model::Text;
use swimos_recon::parser::{AsyncParseError, ParseError, RecognizerDecoder};
use swimos_recon::print_recon_compact;
use swimos_utilities::encoding::BytesStr;
use thiserror::Error;
use tokio_util::codec::{Decoder, Encoder};
use uuid::Uuid;

#[cfg(test)]
mod tests;

/// Operations that can be performed on an agent.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Operation<T> {
    Link,
    Sync,
    Unlink,
    Command(T),
}

impl<T> Operation<T> {
    pub fn is_command(&self) -> bool {
        matches!(self, Operation::Command(_))
    }
}

/// Notifications that can be produced by an agent.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Notification<T, U> {
    Linked,
    Synced,
    Unlinked(Option<U>),
    Event(T),
}

impl<T, U> Notification<T, U>
where
    T: AsRef<[u8]>,
    U: AsRef<[u8]>,
{
    pub fn debug_formatter(&self) -> impl Debug + '_ {
        NotificationDebugFormatter(self)
    }
}

struct NotificationDebugFormatter<'a, T, U>(&'a Notification<T, U>);

impl<'a, T, U> Debug for NotificationDebugFormatter<'a, T, U>
where
    T: AsRef<[u8]>,
    U: AsRef<[u8]>,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.0 {
            Notification::Linked => write!(f, "Linked"),
            Notification::Synced => write!(f, "Synced"),
            Notification::Unlinked(msg) => {
                if let Some(msg) = msg {
                    if let Ok(msg_str) = std::str::from_utf8(msg.as_ref()) {
                        f.debug_tuple("Unlinked")
                            .field(&format!("Str[{:}]", msg_str))
                            .finish()
                    } else {
                        f.debug_tuple("Unlinked")
                            .field(&format!("Bytes[{:?}]", msg.as_ref()))
                            .finish()
                    }
                } else {
                    f.debug_tuple("Unlinked").field(&"Absent").finish()
                }
            }
            Notification::Event(body) => {
                if let Ok(body_str) = std::str::from_utf8(body.as_ref()) {
                    f.debug_tuple("Event")
                        .field(&format!("Str[{}]", body_str))
                        .finish()
                } else {
                    f.debug_tuple("Event")
                        .field(&format!("Bytes[{:?}]", body.as_ref()))
                        .finish()
                }
            }
        }
    }
}

/// Type of messages that can be sent to an agent/from a downlink..
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RequestMessage<P, T> {
    pub origin: Uuid,
    pub path: RelativeAddress<P>,
    pub envelope: Operation<T>,
}

/// Type of messages that can be sent by an agent/received by a downlink.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ResponseMessage<P, T, U> {
    pub origin: Uuid,
    pub path: RelativeAddress<P>,
    pub envelope: Notification<T, U>,
}

impl<P, T> RequestMessage<P, T> {
    pub fn link(source: Uuid, path: RelativeAddress<P>) -> Self {
        RequestMessage {
            origin: source,
            path,
            envelope: Operation::Link,
        }
    }

    pub fn sync(source: Uuid, path: RelativeAddress<P>) -> Self {
        RequestMessage {
            origin: source,
            path,
            envelope: Operation::Sync,
        }
    }

    pub fn unlink(source: Uuid, path: RelativeAddress<P>) -> Self {
        RequestMessage {
            origin: source,
            path,
            envelope: Operation::Unlink,
        }
    }

    pub fn command(source: Uuid, path: RelativeAddress<P>, body: T) -> Self {
        RequestMessage {
            origin: source,
            path,
            envelope: Operation::Command(body),
        }
    }
}

impl<P, T, U> ResponseMessage<P, T, U> {
    pub fn linked(target: Uuid, path: RelativeAddress<P>) -> Self {
        ResponseMessage {
            origin: target,
            path,
            envelope: Notification::Linked,
        }
    }

    pub fn synced(target: Uuid, path: RelativeAddress<P>) -> Self {
        ResponseMessage {
            origin: target,
            path,
            envelope: Notification::Synced,
        }
    }

    pub fn unlinked(target: Uuid, path: RelativeAddress<P>, body: Option<U>) -> Self {
        ResponseMessage {
            origin: target,
            path,
            envelope: Notification::Unlinked(body),
        }
    }

    pub fn event(target: Uuid, path: RelativeAddress<P>, body: T) -> Self {
        ResponseMessage {
            origin: target,
            path,
            envelope: Notification::Event(body),
        }
    }
}

/// An request message where the body is uninterpreted (represented as raw bytes).
pub type RawRequestMessage<'a, P> = RequestMessage<P, &'a [u8]>;
/// A request message where the path and body are taken directly from the input buffer, without copying or interpretation.
pub type BytesRequestMessage = RequestMessage<BytesStr, Bytes>;
/// A response message where the path and body are taken directly from the input buffer, without copying or interpretation.
pub type BytesResponseMessage = ResponseMessage<BytesStr, Bytes, Bytes>;

/// Tokio [`Encoder`] to encode a [`RawRequestMessage`] as a byte stream.
#[derive(Debug, Default)]
pub struct RawRequestMessageEncoder;

/// Tokio [`Encoder`] to encode a raw response message as a byte stream.
#[derive(Debug, Default)]
pub struct RawResponseMessageEncoder;

#[derive(Debug)]
/// Tokio [`Encoder`] to encode an [`ResponseMessage`] as a byte stream.
pub struct ResponseMessageEncoder;

const OP_SHIFT: usize = 61;
const OP_MASK: u64 = 0b111 << OP_SHIFT;

const LINK: u64 = 0b000;
const SYNC: u64 = 0b001;
const UNLINK: u64 = 0b010;
const COMMAND: u64 = 0b011;

const LINKED: u64 = 0b100;
const SYNCED: u64 = 0b101;
const UNLINKED: u64 = 0b110;
const EVENT: u64 = 0b111;

impl<'a, P, B> Encoder<&'a RequestMessage<P, B>> for RawRequestMessageEncoder
where
    P: AsRef<str>,
    B: AsRef<[u8]>,
{
    type Error = std::io::Error;

    fn encode(
        &mut self,
        item: &'a RequestMessage<P, B>,
        dst: &mut BytesMut,
    ) -> Result<(), Self::Error> {
        let RequestMessage {
            origin: source,
            path: RelativeAddress { node, lane },
            envelope,
        } = item;
        let node_str = node.as_ref();
        let lane_str = lane.as_ref();
        dst.reserve(HEADER_INIT_LEN + lane_str.len() + node_str.len());
        dst.put_u128(source.as_u128());
        let node_len = u32::try_from(node_str.len()).expect("Node name to long.");
        let lane_len = u32::try_from(lane_str.len()).expect("Lane name to long.");
        dst.put_u32(node_len);
        dst.put_u32(lane_len);
        match envelope {
            Operation::Link => {
                dst.put_u64(LINK << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Operation::Sync => {
                dst.put_u64(SYNC << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Operation::Unlink => {
                dst.put_u64(UNLINK << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Operation::Command(body) => {
                let body_bytes = body.as_ref();
                let body_len = body_bytes.len() as u64;
                if body_len & OP_MASK != 0 {
                    panic!("Body too large.")
                }
                dst.put_u64(body_len | (COMMAND << OP_SHIFT));
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
                dst.reserve(body_bytes.len());
                dst.put_slice(body_bytes);
            }
        }
        Ok(())
    }
}

impl<P, B> Encoder<RequestMessage<P, B>> for RawRequestMessageEncoder
where
    P: AsRef<str>,
    B: AsRef<[u8]>,
{
    type Error = std::io::Error;

    fn encode(
        &mut self,
        item: RequestMessage<P, B>,
        dst: &mut BytesMut,
    ) -> Result<(), Self::Error> {
        self.encode(&item, dst)
    }
}

impl<'a, P, B1, B2> Encoder<&'a ResponseMessage<P, B1, B2>> for RawResponseMessageEncoder
where
    P: AsRef<str>,
    B1: AsRef<[u8]>,
    B2: AsRef<[u8]>,
{
    type Error = std::io::Error;

    fn encode(
        &mut self,
        item: &'a ResponseMessage<P, B1, B2>,
        dst: &mut BytesMut,
    ) -> Result<(), Self::Error> {
        let ResponseMessage {
            origin: source,
            path: RelativeAddress { node, lane },
            envelope,
        } = item;
        let node_str = node.as_ref();
        let lane_str = lane.as_ref();
        dst.reserve(HEADER_INIT_LEN + lane_str.len() + node_str.len());
        dst.put_u128(source.as_u128());
        let node_len = u32::try_from(node_str.len()).expect("Node name to long.");
        let lane_len = u32::try_from(lane_str.len()).expect("Lane name to long.");
        dst.put_u32(node_len);
        dst.put_u32(lane_len);
        match envelope {
            Notification::Linked => {
                dst.put_u64(LINKED << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Notification::Synced => {
                dst.put_u64(SYNCED << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Notification::Unlinked(body) => {
                let body_len = body.as_ref().map(|b| b.as_ref().len()).unwrap_or_default();
                dst.put_u64(body_len as u64 | (UNLINKED << OP_SHIFT));
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
                dst.reserve(body_len);
                if let Some(body) = body {
                    dst.put_slice(body.as_ref());
                }
            }
            Notification::Event(body) => {
                let body_bytes = body.as_ref();
                let body_len = body_bytes.len() as u64;
                if body_len & OP_MASK != 0 {
                    panic!("Body too large.")
                }
                dst.put_u64(body_len | (EVENT << OP_SHIFT));
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
                dst.reserve(body_bytes.len());
                dst.put_slice(body_bytes);
            }
        }
        Ok(())
    }
}

impl<P, B1, B2> Encoder<ResponseMessage<P, B1, B2>> for RawResponseMessageEncoder
where
    P: AsRef<str>,
    B1: AsRef<[u8]>,
    B2: AsRef<[u8]>,
{
    type Error = std::io::Error;

    fn encode(
        &mut self,
        item: ResponseMessage<P, B1, B2>,
        dst: &mut BytesMut,
    ) -> Result<(), Self::Error> {
        self.encode(&item, dst)
    }
}

const RESERVE_INIT: usize = 256;
const RESERVE_MULT: usize = 2;

impl<P, T, U> Encoder<ResponseMessage<P, T, U>> for ResponseMessageEncoder
where
    P: AsRef<str>,
    T: StructuralWritable,
    U: AsRef<[u8]>,
{
    type Error = std::io::Error;

    fn encode(
        &mut self,
        item: ResponseMessage<P, T, U>,
        dst: &mut BytesMut,
    ) -> Result<(), Self::Error> {
        let ResponseMessage {
            origin: source,
            path: RelativeAddress { node, lane },
            envelope,
        } = item;
        let node_str = node.as_ref();
        let lane_str = lane.as_ref();
        dst.reserve(HEADER_INIT_LEN + node_str.len() + lane_str.len());
        dst.put_u128(source.as_u128());
        let node_len = u32::try_from(node_str.len()).expect("Node name to long.");
        let lane_len = u32::try_from(lane_str.len()).expect("Lane name to long.");
        dst.put_u32(node_len);
        dst.put_u32(lane_len);
        match envelope {
            Notification::Linked => {
                dst.put_u64(LINKED << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Notification::Synced => {
                dst.put_u64(SYNCED << OP_SHIFT);
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
            }
            Notification::Unlinked(body) => {
                let body_len = body.as_ref().map(|b| b.as_ref().len()).unwrap_or_default();
                dst.put_u64(body_len as u64 | (UNLINKED << OP_SHIFT));
                dst.put_slice(node_str.as_bytes());
                dst.put_slice(lane_str.as_bytes());
                dst.reserve(body_len);
                if let Some(body) = body {
                    dst.put_slice(body.as_ref());
                }
            }
            Notification::Event(body) => {
                put_with_body(node.as_ref(), lane.as_ref(), EVENT, &body, dst);
            }
        }
        Ok(())
    }
}

fn put_with_body<T: StructuralWritable>(
    node: &str,
    lane: &str,
    code: u64,
    body: &T,
    dst: &mut BytesMut,
) {
    let body_len_offset = dst.remaining();
    dst.put_u64(0);
    dst.put_slice(node.as_bytes());
    dst.put_slice(lane.as_bytes());
    let body_offset = dst.remaining();

    let mut next_res = RESERVE_INIT.max(dst.remaining_mut().saturating_mul(RESERVE_MULT));
    loop {
        if write!(dst, "{}", print_recon_compact(body)).is_err() {
            dst.truncate(body_offset);
            dst.reserve(next_res);
            next_res = next_res.saturating_mul(RESERVE_MULT);
        } else {
            break;
        }
    }
    let body_len = (dst.remaining() - body_offset) as u64;
    if body_len & OP_MASK != 0 {
        panic!("Body too large.")
    }
    let mut rewound = &mut dst.as_mut()[body_len_offset..];
    rewound.put_u64(body_len | (code << OP_SHIFT));
}

#[derive(Debug)]
enum RequestState<P, T> {
    ReadingHeader,
    ReadingBody {
        source: Uuid,
        path: RelativeAddress<P>,
        remaining: usize,
    },
    AfterBody {
        message: Option<RequestMessage<P, T>>,
        remaining: usize,
    },
    Discarding {
        error: Option<AsyncParseError>,
        remaining: usize,
    },
}

/// Tokio [`Decoder`] that can read an [`RequestMessage`] from a stream of bytes, using a
/// [`RecognizerDecoder`] to interpret the body.
pub struct RequestMessageDecoder<T, R> {
    state: RequestState<Text, T>,
    recognizer: RecognizerDecoder<R>,
}

impl<T: Debug, R> Debug for RequestMessageDecoder<T, R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RequestMessageDecoder")
            .field("state", &self.state)
            .field("recognizer", &"...")
            .finish()
    }
}

/// Tokio [`Decoder`] that can read an raw response message from a stream of bytes.
#[derive(Default, Debug, Clone, Copy)]
pub struct RawResponseMessageDecoder;

/// Tokio [`Decoder`] that can read an [`RawRequestMessage`] from a stream of bytes.
#[derive(Default, Debug, Clone, Copy)]
pub struct RawRequestMessageDecoder;

impl<T, R> RequestMessageDecoder<T, R> {
    pub fn new(recognizer: R) -> Self {
        RequestMessageDecoder {
            state: RequestState::ReadingHeader,
            recognizer: RecognizerDecoder::new(recognizer),
        }
    }
}

const HEADER_INIT_LEN: usize = 32;

/// Error type for the protocol decoders.
#[derive(Error, Debug)]
pub enum MessageDecodeError {
    /// There was an IO error reading from the channel.
    #[error("Error reading from the source: {0}")]
    Io(#[from] std::io::Error),
    /// A lane name in a frame contained invalid UTF-8.
    #[error("Invalid lane name: {0}")]
    LaneName(#[from] Utf8Error),
    /// The kind of a frame was invalid.
    #[error("Unexpecetd message tag code: {0}")]
    UnexpectedCode(u64),
    /// The body of a frame could not be deserialized.
    #[error("Invalid message body: {0}")]
    Body(#[from] AsyncParseError),
}

impl From<ReadError> for MessageDecodeError {
    fn from(e: ReadError) -> Self {
        MessageDecodeError::Body(AsyncParseError::Parser(ParseError::Structure(e)))
    }
}

impl MessageDecodeError {
    fn incomplete() -> Self {
        MessageDecodeError::Body(AsyncParseError::Parser(ParseError::Structure(
            ReadError::IncompleteRecord,
        )))
    }
}

impl<T, R> Decoder for RequestMessageDecoder<T, R>
where
    R: Recognizer<Target = T>,
{
    type Item = RequestMessage<Text, T>;
    type Error = MessageDecodeError;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        let RequestMessageDecoder {
            state, recognizer, ..
        } = self;
        loop {
            match state {
                RequestState::ReadingHeader => {
                    if src.remaining() < HEADER_INIT_LEN {
                        src.reserve(HEADER_INIT_LEN);
                        break Ok(None);
                    }
                    let mut header = &src.as_ref()[0..HEADER_INIT_LEN];
                    let source = header.get_u128();
                    let id = Uuid::from_u128(source);
                    let node_len = header.get_u32() as usize;
                    let lane_len = header.get_u32() as usize;
                    let body_len_and_tag = header.get_u64();
                    let tag = (body_len_and_tag & OP_MASK) >> OP_SHIFT;
                    if src.remaining() < HEADER_INIT_LEN + node_len + lane_len {
                        src.reserve(node_len + lane_len);
                        break Ok(None);
                    }
                    src.advance(HEADER_INIT_LEN);
                    let node = Text::new(std::str::from_utf8(&src.as_ref()[0..node_len])?);
                    src.advance(node_len);
                    let lane = Text::new(std::str::from_utf8(&src.as_ref()[0..lane_len])?);
                    src.advance(lane_len);
                    let path = RelativeAddress::new(node, lane);
                    match tag {
                        LINK => {
                            break Ok(Some(RequestMessage {
                                origin: id,
                                path,
                                envelope: Operation::Link,
                            }));
                        }
                        SYNC => {
                            break Ok(Some(RequestMessage {
                                origin: id,
                                path,
                                envelope: Operation::Sync,
                            }));
                        }
                        UNLINK => {
                            break Ok(Some(RequestMessage {
                                origin: id,
                                path,
                                envelope: Operation::Unlink,
                            }));
                        }
                        COMMAND => {
                            let body_len = (body_len_and_tag & !OP_MASK) as usize;
                            *state = RequestState::ReadingBody {
                                source: id,
                                path,
                                remaining: body_len,
                            };
                        }
                        _ => {
                            break Err(MessageDecodeError::UnexpectedCode(tag));
                        }
                    }
                }
                RequestState::ReadingBody {
                    source,
                    path,
                    remaining,
                } => {
                    let to_split = (*remaining).min(src.remaining());
                    let rem = src.split_off(to_split);
                    let buf_remaining = src.remaining();
                    let end_of_message = *remaining <= buf_remaining;
                    let decode_result = recognizer.decode(src);
                    let new_remaining = src.remaining();
                    let consumed = buf_remaining - new_remaining;
                    *remaining -= consumed;
                    match decode_result {
                        Ok(Some(result)) => {
                            src.unsplit(rem);
                            *state = RequestState::AfterBody {
                                message: Some(RequestMessage {
                                    origin: *source,
                                    path: std::mem::take(path),
                                    envelope: Operation::Command(result),
                                }),
                                remaining: *remaining,
                            }
                        }
                        Ok(None) => {
                            if end_of_message {
                                let eof_result = recognizer.decode_eof(src)?;
                                let final_remaining = src.remaining();
                                let consumed = new_remaining - final_remaining;
                                *remaining -= consumed;
                                src.unsplit(rem);
                                let result = if let Some(result) = eof_result {
                                    Ok(Some(RequestMessage {
                                        origin: *source,
                                        path: std::mem::take(path),
                                        envelope: Operation::Command(result),
                                    }))
                                } else {
                                    Err(MessageDecodeError::incomplete())
                                };
                                *state = RequestState::ReadingHeader;
                                break result;
                            } else {
                                break Ok(None);
                            }
                        }
                        Err(e) => {
                            *remaining -= new_remaining;
                            src.unsplit(rem);
                            src.advance(new_remaining);
                            if *remaining == 0 {
                                *state = RequestState::ReadingHeader;
                                break Err(e.into());
                            } else {
                                *state = RequestState::Discarding {
                                    error: Some(e),
                                    remaining: *remaining,
                                }
                            }
                        }
                    }
                }
                RequestState::AfterBody { message, remaining } => {
                    if src.remaining() >= *remaining {
                        src.advance(*remaining);
                        let result = message.take();
                        *state = RequestState::ReadingHeader;
                        break Ok(result);
                    } else {
                        *remaining -= src.remaining();
                        src.clear();
                        break Ok(None);
                    }
                }
                RequestState::Discarding { error, remaining } => {
                    if src.remaining() >= *remaining {
                        src.advance(*remaining);
                        let err = error.take().unwrap_or(AsyncParseError::UnconsumedInput);
                        *state = RequestState::ReadingHeader;
                        break Err(err.into());
                    } else {
                        *remaining -= src.remaining();
                        src.clear();
                        break Ok(None);
                    }
                }
            }
        }
    }
}

impl Decoder for RawResponseMessageDecoder {
    type Item = ResponseMessage<BytesStr, Bytes, Bytes>;
    type Error = std::io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        if src.remaining() < HEADER_INIT_LEN {
            src.reserve(HEADER_INIT_LEN - src.remaining() + 1);
            return Ok(None);
        }
        let mut header = &src.as_ref()[0..HEADER_INIT_LEN];
        let target = header.get_u128();
        let target = Uuid::from_u128(target);
        let node_len = header.get_u32() as usize;
        let lane_len = header.get_u32() as usize;
        let body_len_and_tag = header.get_u64();
        let body_len = (body_len_and_tag & !OP_MASK) as usize;
        let required = HEADER_INIT_LEN + node_len + lane_len + body_len;
        if src.remaining() < required {
            src.reserve(required - src.remaining());
            return Ok(None);
        }
        src.advance(HEADER_INIT_LEN);
        let node_bytes = src.split_to(node_len).freeze();
        let node = BytesStr::try_from(node_bytes)
            .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))?;

        let lane_bytes = src.split_to(lane_len).freeze();
        let lane = BytesStr::try_from(lane_bytes)
            .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))?;

        let path = RelativeAddress::new(node, lane);
        let tag = (body_len_and_tag & OP_MASK) >> OP_SHIFT;
        match tag {
            LINKED => Ok(Some(BytesResponseMessage::linked(target, path))),
            SYNCED => Ok(Some(BytesResponseMessage::synced(target, path))),
            UNLINKED => {
                let body = if body_len == 0 {
                    None
                } else {
                    Some(src.split_to(body_len).freeze())
                };
                Ok(Some(BytesResponseMessage::unlinked(target, path, body)))
            }
            _ => {
                let body = src.split_to(body_len).freeze();
                Ok(Some(BytesResponseMessage::event(target, path, body)))
            }
        }
    }
}

impl Decoder for RawRequestMessageDecoder {
    type Item = RequestMessage<BytesStr, Bytes>;
    type Error = std::io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        if src.remaining() < HEADER_INIT_LEN {
            src.reserve(HEADER_INIT_LEN - src.remaining() + 1);
            return Ok(None);
        }
        let mut header = &src.as_ref()[0..HEADER_INIT_LEN];
        let origin = header.get_u128();
        let origin = Uuid::from_u128(origin);
        let node_len = header.get_u32() as usize;
        let lane_len = header.get_u32() as usize;
        let body_len_and_tag = header.get_u64();
        let body_len = (body_len_and_tag & !OP_MASK) as usize;
        let required = HEADER_INIT_LEN + node_len + lane_len + body_len;
        if src.remaining() < required {
            src.reserve(required);
            return Ok(None);
        }
        src.advance(HEADER_INIT_LEN);
        let node_bytes = src.split_to(node_len).freeze();
        let node = BytesStr::try_from(node_bytes)
            .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))?;

        let lane_bytes = src.split_to(lane_len).freeze();
        let lane = BytesStr::try_from(lane_bytes)
            .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))?;

        let path = RelativeAddress::new(node, lane);
        let tag = (body_len_and_tag & OP_MASK) >> OP_SHIFT;
        match tag {
            LINK => Ok(Some(RequestMessage::link(origin, path))),
            SYNC => Ok(Some(RequestMessage::sync(origin, path))),
            UNLINK => Ok(Some(RequestMessage::unlink(origin, path))),
            _ => {
                let body = src.split_to(body_len).freeze();
                Ok(Some(RequestMessage::command(origin, path, body)))
            }
        }
    }
}
