Apply backpressure in pushers, portmap

examples/server-jobs-example/.env

@@ -1 +1 @@
-RUST_LOG=server_jobs_example,jobs_server_tokio=info
+RUST_LOG=info

jobs-server-tokio/Cargo.toml

@@ -12,7 +12,7 @@ serde = "1.0"
 serde_json = "1.0"
 tokio = "0.1"
 tokio-threadpool = "0.1"
-tokio-zmq = "0.5"
+tokio-zmq = "0.6.1"
 zmq = "0.8"
 
 [dependencies.jobs-core]

jobs-server-tokio/src/server/portmap.rs

@@ -40,7 +40,7 @@ impl PortMapConfig {
             config: _,
         } = self;
 
-        let (sink, stream) = rep.sink_stream().split();
+        let (sink, stream) = rep.sink_stream(1).split();
 
         let fut = stream
             .from_err::<Error>()

jobs-server-tokio/src/server/push.rs

@@ -57,7 +57,7 @@ impl PushConfig {
             .from_err()
             .and_then(move |_| dequeue_jobs(storage.clone(), queue.clone()))
             .flatten()
-            .forward(pusher.sink())
+            .forward(pusher.sink(25))
             .map(move |_| {
                 info!(
                     "Pusher for queue {} is shutting down",
@@ -137,6 +137,7 @@ impl ResetPushConfig {
 
     fn build(self) -> impl Future<Item = (), Error = Error> {
         lazy(|| {
+            info!("Building and spawning new server");
             let pusher = Push::builder(self.config.context.clone())
                 .bind(&self.address)
                 .build()?;

jobs-server-tokio/src/worker/config.rs

@@ -57,7 +57,7 @@ impl Worker {
             .stream()
             .from_err::<Error>()
             .and_then(move |multipart| wrap_processing(multipart, &processors))
-            .forward(push.sink())
+            .forward(push.sink(2))
             .map(move |_| info!("worker for queue {} is shutting down", queue))
             .map_err(|e| {
                 error!("Error processing job, {}", e);