meilies-client 0.1.1

A TCP client for MeiliES
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

use std::net::SocketAddr;
use std::{fmt, io};

use futures::{Future, Stream, Sink};
use tokio_retry::Retry;
use log::warn;
use meilies::stream::{StreamName, EventNumber, EventData, EventName};
use meilies::reqresp::{Request, RequestMsgError};
use meilies::reqresp::{Response, ResponseMsgError};

use crate::steel_connection::retry_strategy;
use super::{connect, SteelConnection};

/// Open a framed paired connection with a server.
pub fn paired_connect(
    addr: SocketAddr
) -> impl Future<Item=PairedConnection, Error=tokio_retry::Error<io::Error>>
{
    PairedConnection::connect(addr)
}

/// A paired connection returns a response to each message send, it is sequential.
/// This connection is used to publish events to streams.
pub struct PairedConnection {
    connection: SteelConnection,
}

#[derive(Debug)]
pub enum PairedConnectionError {
    ServerSide(String),
    ConnectionClosed,
    RequestMsgError(RequestMsgError),
    ResponseMsgError(ResponseMsgError),
    InvalidServerResponse(Response),
}

impl fmt::Display for PairedConnectionError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        use PairedConnectionError::*;

        match self {
            ServerSide(error) => write!(f, "server side error: {}", error),
            ConnectionClosed => write!(f, "connection closed"),
            RequestMsgError(error) => write!(f, "invalid Request: {}", error),
            ResponseMsgError(error) => write!(f, "invalid Response received: {}", error),
            InvalidServerResponse(response) => {
                write!(f, "invalid server response received: {:?}", response)
            },
        }
    }
}

impl PairedConnection {
    /// Open a framed paired connection with a server.
    pub fn connect(addr: SocketAddr) -> impl Future<Item=PairedConnection, Error=tokio_retry::Error<io::Error>> {
        Retry::spawn(retry_strategy(), move || {
            warn!("Connecting to {}", addr);
            connect(&addr)
                .map(move |connection| {
                    let connection = SteelConnection::new(addr, connection);
                    PairedConnection { connection }
                })
        })
    }

    /// Publish an event to a stream, specifying the event name and data.
    pub fn publish(
        self,
        stream: StreamName,
        event_name: EventName,
        event_data: EventData,
    ) -> impl Future<Item=PairedConnection, Error=PairedConnectionError>
    {
        use PairedConnectionError::*;

        let command = Request::Publish { stream, event_name, event_data };

        self.connection
            .send(command)
            .map_err(RequestMsgError)
            .and_then(|framed| framed.into_future().map_err(|(e, _)| ResponseMsgError(e)))
            .and_then(|(first, connection)| {
                match first.ok_or(ConnectionClosed)? {
                    Ok(Response::Ok) => Ok(PairedConnection { connection }),
                    Ok(response) => Err(InvalidServerResponse(response)),
                    Err(error) => Err(ServerSide(error)),
                }
            })
    }

    /// Request the last event number that the stream is at.
    ///
    /// Returns `None` if the stream does not contain any event.
    pub fn last_event_number(
        self,
        stream: StreamName
    ) -> impl Future<Item=(StreamName, Option<EventNumber>, PairedConnection), Error=PairedConnectionError>
    {
        use PairedConnectionError::*;

        let command = Request::LastEventNumber { stream };

        self.connection
            .send(command)
            .map_err(RequestMsgError)
            .and_then(|framed| framed.into_future().map_err(|(e, _)| ResponseMsgError(e)))
            .and_then(|(first, connection)| {
                match first.ok_or(ConnectionClosed)? {
                    Ok(Response::LastEventNumber { stream, number }) => Ok((stream, number, PairedConnection { connection })),
                    Ok(response) => Err(InvalidServerResponse(response)),
                    Err(error) => Err(ServerSide(error)),
                }
            })
    }
}