nitox 0.1.9

Nitox is a `tokio`-based client for NATS.
Documentation 
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

use super::{commands::*, Command, CommandError};
use bytes::Bytes;

/// Abstraction over NATS protocol messages
#[derive(Debug, Clone, PartialEq)]
pub enum Op {
    /// **SERVER** Sent to client after initial TCP/IP connection
    INFO(ServerInfo),
    /// **CLIENT** Sent to server to specify connection information
    CONNECT(ConnectCommand),
    /// **CLIENT** Publish a message to a subject, with optional reply subject
    PUB(PubCommand),
    /// **CLIENT** Subscribe to a subject (or subject wildcard)
    SUB(SubCommand),
    /// **CLIENT** Unsubscribe (or auto-unsubscribe) from subject
    UNSUB(UnsubCommand),
    /// **SERVER** Delivers a message payload to a subscriber
    MSG(Message),
    /// **BOTH** PING keep-alive message
    PING,
    /// **BOTH** PONG keep-alive message
    PONG,
    /// **SERVER** Acknowledges well-formed protocol message in `verbose` mode
    OK,
    /// **SERVER** Indicates a protocol error. May cause client disconnect.
    ERR(ServerError),
}

macro_rules! op_from_cmd {
    ($buf:ident, $cmd:path, $op:path) => {{
        use protocol::CommandError;

        match $cmd(&$buf) {
            Ok(c) => Ok($op(c)),
            Err(CommandError::IncompleteCommandError) => return Err(CommandError::IncompleteCommandError),
            Err(e) => return Err(e.into()),
        }
    }};
}

impl Op {
    /// Transforms the OP into a byte slice
    pub fn into_bytes(self) -> Result<Bytes, CommandError> {
        Ok(match self {
            Op::INFO(si) => si.into_vec()?,
            Op::CONNECT(con) => con.into_vec()?,
            Op::PUB(pc) => pc.into_vec()?,
            Op::SUB(sc) => sc.into_vec()?,
            Op::UNSUB(uc) => uc.into_vec()?,
            Op::MSG(msg) => msg.into_vec()?,
            Op::PING => "PING\r\n".into(),
            Op::PONG => "PONG\r\n".into(),
            Op::OK => "+OK\r\n".into(),
            Op::ERR(se) => format!("-ERR {}\r\n", se).as_bytes().into(),
        })
    }

    /// Tries to parse from a pair of command name and whole buffer
    pub fn from_bytes(cmd_name: &[u8], buf: &[u8]) -> Result<Self, CommandError> {
        match cmd_name {
            ServerInfo::CMD_NAME => op_from_cmd!(buf, ServerInfo::try_parse, Op::INFO),
            ConnectCommand::CMD_NAME => op_from_cmd!(buf, ConnectCommand::try_parse, Op::CONNECT),
            Message::CMD_NAME => op_from_cmd!(buf, Message::try_parse, Op::MSG),
            PubCommand::CMD_NAME => op_from_cmd!(buf, PubCommand::try_parse, Op::PUB),
            SubCommand::CMD_NAME => op_from_cmd!(buf, SubCommand::try_parse, Op::SUB),
            UnsubCommand::CMD_NAME => op_from_cmd!(buf, UnsubCommand::try_parse, Op::UNSUB),
            b"PING" => {
                if buf == b"PING\r\n" {
                    Ok(Op::PING)
                } else {
                    Err(CommandError::IncompleteCommandError)
                }
            }
            b"PONG" => {
                if buf == b"PONG\r\n" {
                    Ok(Op::PONG)
                } else {
                    Err(CommandError::IncompleteCommandError)
                }
            }
            b"+OK" => {
                if buf == b"+OK\r\n" {
                    Ok(Op::OK)
                } else {
                    Err(CommandError::IncompleteCommandError)
                }
            }
            b"-ERR" => {
                if &buf[buf.len() - 2..] == b"\r\n" {
                    Ok(Op::ERR(ServerError::from(String::from_utf8(buf[1..].to_vec())?)))
                } else {
                    Err(CommandError::IncompleteCommandError)
                }
            }
            _ => {
                if buf.len() > 7 {
                    Err(CommandError::CommandNotFoundOrSupported)
                } else {
                    Err(CommandError::IncompleteCommandError)
                }
            }
        }
    }
}

// TODO: Write tests