tokio-zmq 0.10.1

Provides Futures abstractions for ZeroMQ on the Tokio event-loop
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
130
131
132
133
134
135
136
137
138
139
140
141

/*
 * This file is part of Tokio ZMQ.
 *
 * Copyright © 2018 Riley Trautman
 *
 * Tokio ZMQ is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Tokio ZMQ is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Tokio ZMQ.  If not, see <http://www.gnu.org/licenses/>.
 */

extern crate env_logger;
extern crate futures;
extern crate log;
extern crate tokio;
extern crate tokio_zmq;
extern crate zmq;

use std::{sync::Arc, thread, time::Duration};

use futures::{stream::iter_ok, Future, Sink, Stream};
use tokio_zmq::{prelude::*, Error, Multipart, Pub, Rep, Req, Sub};

// On my quad-core i7, if I run with too many threads, the context switching takes too long and
// some messages get dropped. 2 subscribers can properly retrieve 1 million messages each, though.
//
// For the nice results, lower the messages and increas the subscribers.
const SUBSCRIBERS: usize = 10;
const MESSAGES: usize = 1_000;

struct Stop;

impl EndHandler for Stop {
    fn should_stop(&mut self, item: &Multipart) -> bool {
        if let Some(msg) = item.get(0) {
            if let Some(msg) = msg.as_str() {
                if msg == "END" {
                    return true;
                }
            }
        }

        false
    }
}

fn publisher_thread() {
    let ctx = Arc::new(zmq::Context::new());

    let publisher_fut = Pub::builder(Arc::clone(&ctx)).bind("tcp://*:5561").build();

    let syncservice_fut = Rep::builder(ctx).bind("tcp://*:5562").build();

    println!("Waiting for subscribers");

    let runner = publisher_fut
        .join(syncservice_fut)
        .and_then(|(publisher, syncservice)| {
            let (sync_sink, sync_stream) = syncservice.sink_stream(25).split();

            iter_ok(0..SUBSCRIBERS)
                .zip(sync_stream)
                .map(|(_, _)| zmq::Message::from("").into())
                .forward(sync_sink)
                .and_then(move |_| {
                    println!("Broadcasting message");

                    iter_ok(0..MESSAGES)
                        .map(|_| zmq::Message::from("Rhubarb").into())
                        .forward(publisher.sink(25))
                })
                .and_then(|(_stream, sink)| {
                    let msg = zmq::Message::from("END");

                    sink.send(msg.into())
                })
        });

    tokio::run(runner.map(|_| ()).or_else(|e| {
        println!("Error in publisher: {:?}", e);
        Ok(())
    }));
}

fn subscriber_thread() {
    let ctx = Arc::new(zmq::Context::new());

    let subscriber_fut = Sub::builder(Arc::clone(&ctx))
        .connect("tcp://localhost:5561")
        .filter(b"")
        .build();

    let syncclient_fut = Req::builder(ctx).connect("tcp://localhost:5562").build();

    let msg = zmq::Message::from("");

    let runner = subscriber_fut
        .join(syncclient_fut)
        .and_then(|(subscriber, syncclient)| {
            syncclient
                .send(msg.into())
                .and_then(|syncclient| syncclient.recv())
                .and_then(move |_| {
                    subscriber
                        .stream()
                        .with_end_handler(Stop)
                        .fold(0, |counter, _| Ok(counter + 1) as Result<usize, Error>)
                        .and_then(|total| {
                            println!("Received {} updates", total);
                            Ok(())
                        })
                })
        });

    tokio::run(runner.map(|_| ()).or_else(|e| {
        println!("Error in subscriber: {:?}", e);
        Ok(())
    }));
}

fn main() {
    let mut threads = Vec::new();
    threads.push(thread::spawn(publisher_thread));

    for _ in 0..SUBSCRIBERS {
        thread::sleep(Duration::from_millis(400));
        threads.push(thread::spawn(subscriber_thread));
    }

    for thread in threads {
        thread.join().unwrap();
    }
}