1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

//! Handlers for converting byte to message

#[cfg(not(feature = "sans-io"))]
use async_trait::async_trait;
use bytes::BytesMut;

use crate::channel::{Handler, InboundContext, InboundHandler, OutboundContext, OutboundHandler};

mod line_based_frame_decoder;
mod tagged;

pub use line_based_frame_decoder::{LineBasedFrameDecoder, TerminatorType};
pub use tagged::TaggedByteToMessageCodec;

/// This trait allows for decoding messages.
pub trait MessageDecoder {
    /// Decodes byte buffer to message buffer
    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<BytesMut>, std::io::Error>;
}

struct ByteToMessageDecoder {
    transport_active: bool,
    message_decoder: Box<dyn MessageDecoder + Send + Sync>,
}

struct ByteToMessageEncoder;

/// A Byte to Message Codec handler that reads with input of BytesMut and output of BytesMut,
/// or writes with input of BytesMut and output of BytesMut
pub struct ByteToMessageCodec {
    decoder: ByteToMessageDecoder,
    encoder: ByteToMessageEncoder,
}

impl ByteToMessageCodec {
    /// Creates a new ByteToMessageCodec handler
    pub fn new(message_decoder: Box<dyn MessageDecoder + Send + Sync>) -> Self {
        Self {
            decoder: ByteToMessageDecoder {
                transport_active: false,
                message_decoder,
            },
            encoder: ByteToMessageEncoder {},
        }
    }
}

#[cfg(not(feature = "sans-io"))]
#[async_trait]
impl InboundHandler for ByteToMessageDecoder {
    type Rin = BytesMut;
    type Rout = Self::Rin;

    async fn transport_active(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>) {
        self.transport_active = true;
        ctx.fire_transport_active().await;
    }
    async fn transport_inactive(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>) {
        self.transport_active = false;
        ctx.fire_transport_inactive().await;
    }
    async fn read(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>, mut msg: Self::Rin) {
        while self.transport_active {
            match self.message_decoder.decode(&mut msg) {
                Ok(message) => {
                    if let Some(message) = message {
                        ctx.fire_read(message).await;
                    } else {
                        return;
                    }
                }
                Err(err) => {
                    ctx.fire_read_exception(Box::new(err)).await;
                    return;
                }
            }
        }
    }
}

#[cfg(not(feature = "sans-io"))]
#[async_trait]
impl OutboundHandler for ByteToMessageEncoder {
    type Win = BytesMut;
    type Wout = Self::Win;

    async fn write(&mut self, ctx: &OutboundContext<Self::Win, Self::Wout>, msg: Self::Win) {
        ctx.fire_write(msg).await;
    }
}

#[cfg(feature = "sans-io")]
impl InboundHandler for ByteToMessageDecoder {
    type Rin = BytesMut;
    type Rout = Self::Rin;

    fn transport_active(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>) {
        self.transport_active = true;
        ctx.fire_transport_active();
    }
    fn transport_inactive(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>) {
        self.transport_active = false;
        ctx.fire_transport_inactive();
    }
    fn read(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>, mut msg: Self::Rin) {
        while self.transport_active {
            match self.message_decoder.decode(&mut msg) {
                Ok(message) => {
                    if let Some(message) = message {
                        ctx.fire_read(message);
                    } else {
                        return;
                    }
                }
                Err(err) => {
                    ctx.fire_read_exception(Box::new(err));
                    return;
                }
            }
        }
    }
}

#[cfg(feature = "sans-io")]
impl OutboundHandler for ByteToMessageEncoder {
    type Win = BytesMut;
    type Wout = Self::Win;

    fn write(&mut self, ctx: &OutboundContext<Self::Win, Self::Wout>, msg: Self::Win) {
        ctx.fire_write(msg);
    }
}

impl Handler for ByteToMessageCodec {
    type Rin = BytesMut;
    type Rout = Self::Rin;
    type Win = BytesMut;
    type Wout = Self::Win;

    fn name(&self) -> &str {
        "ByteToMessageCodec"
    }

    fn split(
        self,
    ) -> (
        Box<dyn InboundHandler<Rin = Self::Rin, Rout = Self::Rout>>,
        Box<dyn OutboundHandler<Win = Self::Win, Wout = Self::Wout>>,
    ) {
        (Box::new(self.decoder), Box::new(self.encoder))
    }
}