phoenix_channels/
client.rs

1use std::thread;
2use std::time::Duration;
3use std::sync::{Arc, Mutex};
4use std::sync::mpsc;
5
6use slog;
7use slog_stdlog;
8use slog::Drain;
9
10use websocket::client::ClientBuilder;
11
12use receiver::Receiver;
13use sender::Sender;
14use error::ConnectError;
15use message::Message;
16use error::{JoinError, MessageError};
17
18pub type MessageResult = Result<Message, MessageError>;
19pub type MessageReceiver = mpsc::Receiver<MessageResult>;
20
21
22const PHOENIX_VERSION: &str = "2.0.0";
23
24
25#[derive(Debug)]
26pub enum ClientError {
27    Connect(ConnectError),
28    Join(JoinError),
29    Thread(String),
30}
31
32impl From<ConnectError> for ClientError {
33    fn from(e: ConnectError) -> Self {
34        return ClientError::Connect(e);
35    }
36}
37
38impl From<JoinError> for ClientError {
39    fn from(e: JoinError) -> Self {
40        return ClientError::Join(e);
41    }
42}
43
44
45
46pub fn connect(url: &str, params: Vec<(&str, &str)>, logger: Option<slog::Logger>) -> Result<(Sender, Receiver), ConnectError> {
47    let logger = logger.unwrap_or(slog::Logger::root(slog_stdlog::StdLog.fuse(), o!()));
48
49    // convert the params to a uri component string
50    let mut params_uri: String = "".to_owned();
51    for (k, v) in params {
52        params_uri.push_str(&format!("&{}={}", k, v));
53    }
54
55    // create a phoenix socket url with params expanded and parse it
56    // phoenix socket endpoints always have /websocket appended for the socket route
57    // it also adds the vsn parameter for versioning
58    let addr = format!("{}/websocket?vsn={}{}", url, PHOENIX_VERSION, params_uri);
59    let mut client_builder = ClientBuilder::new(&addr)?;
60
61    let socket_client = client_builder.connect_insecure()?;
62    let (reader, writer) = socket_client.split()?;
63
64    let sender = Sender::new(writer, logger.new(o!("type" => "sender")));
65    let receiver = Receiver::new(reader, logger.new(o!("type" => "receiver")));
66
67    return Ok((sender, receiver));
68}
69
70
71pub struct Client {
72    logger: slog::Logger,
73    sender_ref: Arc<Mutex<Sender>>,
74    heartbeat_handle: thread::JoinHandle<()>,
75    message_processor_handle: thread::JoinHandle<()>,
76}
77
78impl Client {
79    pub fn new(url: &str, params: Vec<(&str, &str)>, logger: Option<slog::Logger>) -> Result<(Client, MessageReceiver), ClientError> {
80        let logger = logger.unwrap_or(slog::Logger::root(slog_stdlog::StdLog.fuse(), o!()));
81        debug!(logger, "creating client"; "url" => url);
82
83        let (sender, receiver) = connect(url, params, Some(logger.clone()))?;
84
85        let (tx, rx) = mpsc::channel();
86
87        let sender_ref = Arc::new(Mutex::new(sender));
88        let heartbeat = Client::keepalive(Arc::clone(&sender_ref));
89        let message_processor = Client::process_messages(receiver, tx);
90
91        let client = Client {
92            logger: logger,
93            sender_ref: sender_ref,
94            heartbeat_handle: heartbeat,
95            message_processor_handle: message_processor,
96        };
97
98        return Ok((client, rx));
99    }
100
101    fn keepalive(sender_ref: Arc<Mutex<Sender>>) -> thread::JoinHandle<()> {
102        return thread::spawn(move || {
103            loop {
104                thread::sleep(Duration::from_secs(30));
105                // if the mutex is poisoned then the whole thread wont work
106                let mut sender = sender_ref.lock().unwrap();
107                sender.heartbeat();
108            }
109        });
110    }
111
112    fn process_messages(receiver: Receiver, sender: mpsc::Sender<MessageResult>) -> thread::JoinHandle<()> {
113        return thread::spawn(move || {
114            for message in MessageIterator::new(receiver) {
115                let result = sender.send(message);
116
117                // exit the thread cleanly if the channel is closed
118                if result.is_err() {
119                    break;
120                }
121            }
122        });
123    }
124
125    pub fn join(&self, channel: &str) -> Result<u32, ClientError> {
126        return match self.sender_ref.lock() {
127            Ok(mut sender) => Ok(sender.join(channel)?),
128            Err(_) => Err(ClientError::Thread(String::from("Cannot join as sender mutex has been poisoned"))),
129        };
130    }
131
132    pub fn join_threads(self) -> thread::Result<()> {
133        self.heartbeat_handle.join()?;
134        self.message_processor_handle.join()?;
135        return Ok(());
136    }
137}
138
139
140pub struct MessageIterator
141{
142    receiver: Receiver,
143}
144
145impl MessageIterator {
146    pub fn new(receiver: Receiver) -> MessageIterator {
147        MessageIterator {
148            receiver: receiver,
149        }
150    }
151}
152
153impl Iterator for MessageIterator {
154    type Item = MessageResult;
155
156    fn next(&mut self) -> Option<Self::Item> {
157        return self.receiver.next();
158    }
159}
phoenix_channels/client.rs

phoenix_channels/
client.rs
