nsqueue 0.1.5

Rust client for the NSQ realtime message processing system
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use std::io;

use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::codec::{Framed};
use tokio_proto::streaming::pipeline::{Frame, ClientProto};

use serde_json::{to_string};

use futures::{Future, Stream, Sink};

use commands::*;
use response::Message;
use codec::NsqCodec;
use config::Config;

/// Protocol definition
pub struct NsqProtocol {
    pub config: Config,
}

impl NsqProtocol {
    pub fn new(config: Config) -> Self {
        NsqProtocol {
            config: config,
        }
    }
}

#[allow(unused_variables)]
impl<T: AsyncRead + AsyncWrite + 'static> ClientProto<T> for NsqProtocol {
    type Request = RequestMessage;
    type RequestBody = RequestMessage;
    type Response = String;
    type ResponseBody = Message;
    
    type Error = io::Error;
    type Transport = Framed<T, NsqCodec>;
    type BindTransport = Box<Future<Item = Self::Transport, Error = io::Error>>;

    fn bind_transport(&self, io: T) -> Self::BindTransport {
        let config = self.config.clone();
        let mut request = RequestMessage::new();
        request.set_protocol_version(commands::VERSION_2);
                
        let codec = NsqCodec {
            decoding_head: true,
        };

        // Send protocol version
        let tst = request.clone();
        let version = Frame::Message {message: tst, body: false };
        let handshake = io.framed(codec).send(version)
        .and_then(move |transport| {
            let mut request = RequestMessage::new();
            request.create_identify_command(config);
            
            // Send IDENTIFY
            let identify = Frame::Message {message: request.clone(), body: false };
            let ch = transport.send(identify)
                .and_then(|transport| transport.into_future().map_err(|(e, _)| e))
                .and_then(|(resp, transport)| {
                    Ok(transport)                      
                });
            ch
        });
        
        Box::new(handshake)
    }
}

#[derive(PartialEq, Debug, Clone)]
pub struct RequestMessage {
    pub version: Option<String>, 
    pub header: Option<String>,
    pub body: Option<String>,
    pub body_messages: Option<Vec<String>>,
}

impl RequestMessage {
    pub fn new() -> RequestMessage {
        RequestMessage {
            version: None,
            header: None,
            body: None,
            body_messages: None,
        }
    }

    pub fn set_protocol_version(&mut self, version: &str) {
        self.version = Some(String::from(version));
    }

    pub fn create_pub_command(&mut self, topic: String, message: String) {
        self.header = Some(format!("{} {}\n", commands::PUB, topic));
        self.body = Some(message);
    }

    pub fn create_mpub_command(&mut self, topic: String, messages: Vec<String>) {
        self.header = Some(format!("{} {}\n", commands::MPUB, topic));
        self.body_messages = Some(messages);
    }

    pub fn create_dpub_command(&mut self, topic: String, message: String, defer_time: i64) {
        self.header = Some(format!("{} {} {}\n", commands::DPUB, topic, defer_time.to_string()));
        self.body = Some(message);
    }

    pub fn create_sub_command(&mut self, topic: String, channel: String) {
        self.header = Some(format!("{} {} {}\n", commands::SUB, topic, channel));
    }

    pub fn create_identify_command(&mut self, config: Config) {
        self.header = Some(format!("{}\n", commands::IDENTIFY));
        // Serialize it to a JSON string.
        self.body = Some(to_string(&config).unwrap());
    }  

    pub fn create_rdy_command(&mut self) {
        self.header = Some(format!("{} 1\n", commands::RDY));
    }  

    pub fn create_fin_command(&mut self, message_id: String) {
        self.header = Some(format!("{} {}\n", commands::FIN, message_id));
    } 

    pub fn create_nop_command(&mut self) {
        self.header = Some(format!("{}\n", commands::NOP));
    }           
}