batch 0.1.1

Distributed task queue library based on RabbitMQ
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

//! Batch client.

use std::iter::FromIterator;

use futures::{future, Future};
use tokio_core::reactor::Handle;

use error::{Error, ErrorKind};
use job::Job;
use rabbitmq::{Exchange, ExchangeBuilder, RabbitmqBroker};

/// A builder to ease the construction of `Client` instances.
#[derive(Debug)]
pub struct ClientBuilder {
    connection_url: String,
    exchanges: Vec<Exchange>,
    handle: Option<Handle>,
}

impl Default for ClientBuilder {
    fn default() -> ClientBuilder {
        ClientBuilder {
            connection_url: "amqp://localhost/%2f".into(),
            exchanges: Vec::new(),
            handle: None,
        }
    }
}

impl ClientBuilder {
    /// Create a new `ClientBuilder` instance.
    pub fn new() -> Self {
        ClientBuilder::default()
    }

    /// Set the URL used to connect to `RabbitMQ`.
    ///
    /// The URL must be a valid AMQP connection URL (ex: `amqp://localhost/%2f`) using either the
    /// `amqp` protocol or the `amqps` protocol.
    ///
    /// # Example
    ///
    /// ```
    /// use batch::ClientBuilder;
    ///
    /// let builder = ClientBuilder::new()
    ///     .connection_url("amqp://guest:guest@localhost:5672/%2f");
    /// ```
    pub fn connection_url(mut self, url: &str) -> Self {
        self.connection_url = url.into();
        self
    }

    /// Add exchanges to be declared when connecting to `RabbitMQ`.
    ///
    /// See `exchange` documentation.
    ///
    /// # Example
    ///
    /// ```
    /// use batch::{exchange, ClientBuilder};
    ///
    /// let exchanges = vec![
    ///     exchange("batch.example"),
    /// ];
    /// let builder = ClientBuilder::new()
    ///     .exchanges(exchanges);
    /// ```
    pub fn exchanges<EIter>(mut self, exchanges: EIter) -> Self
    where
        EIter: IntoIterator<Item = ExchangeBuilder>,
    {
        self.exchanges
            .extend(exchanges.into_iter().map(|e| e.build()));
        self
    }

    /// Set the `Handle` to the Tokio reactor that should be used by the `Worker`.
    ///
    /// # Example
    ///
    /// ```
    /// # extern crate batch;
    /// # extern crate tokio_core;
    /// #
    /// use batch::ClientBuilder;
    /// use tokio_core::reactor::Core;
    ///
    /// # fn main() {
    /// let core = Core::new().unwrap();
    /// let handle = core.handle();
    /// let builder = ClientBuilder::new()
    ///     .handle(handle);
    /// # }
    /// ```
    pub fn handle(mut self, handle: Handle) -> Self {
        self.handle = Some(handle);
        self
    }

    /// Build a new `Client` instance from this builder data.
    pub fn build(self) -> Box<Future<Item = Client, Error = Error>> {
        if self.handle.is_none() {
            return Box::new(future::err(ErrorKind::NoHandle.into()));
        }
        let task = RabbitmqBroker::new_with_handle(
            &self.connection_url,
            self.exchanges,
            vec![],
            self.handle.unwrap(),
        ).and_then(|broker| Ok(Client { broker }));
        Box::new(task)
    }
}

/// The `Client` is responsible for sending tasks to the broker.
#[derive(Debug)]
pub struct Client {
    broker: RabbitmqBroker,
}

impl Client {
    /// Send a job to the client's message broker.
    ///
    /// Once a job is sent to the message broker, it is transmitted to a Worker currently
    /// receiving jobs from the same broker.
    pub(crate) fn send(&self, job: &Job) -> Box<Future<Item = (), Error = Error>> {
        let task = self.broker.send(job);
        Box::new(task)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn assert_send<T: Send>() {}

    fn assert_sync<T: Sync>() {}

    #[test]
    fn test_auto_impl_traits() {
        assert_send::<Client>();
        assert_sync::<Client>();
    }
}