Preserve runtime output stream order

golemfactory · Jan 15, 2021 · 191822a · 191822a
1 parent dbaf759
commit 191822a
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 20 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/exe-unit/runtime-api/Cargo.toml b/exe-unit/runtime-api/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ya-runtime-api"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Golem Factory <contact@golem.network>"]
 edition = "2018"
 

diff --git a/exe-unit/runtime-api/src/server/service.rs b/exe-unit/runtime-api/src/server/service.rs
@@ -85,14 +85,9 @@ where
         async move {
             while let Some(event) = rx.next().await {
                 log::debug!("event: {:?}", event);
-                tokio::task::spawn_local({
-                    let output = output.clone();
-                    async move {
-                        let mut output = output.lock().await;
-                        let r = SinkExt::send(&mut *output, event).await;
-                        log::debug!("sending event done: {:?}", r);
-                    }
-                });
+                let mut output = output.lock().await;
+                let r = SinkExt::send(&mut *output, event).await;
+                log::debug!("sending event done: {:?}", r);
             }
         }
     });
@@ -104,15 +99,13 @@ where
                     Ok(request) => {
                         let service = service.clone();
                         let output = output.clone();
-                        tokio::task::spawn_local(async move {
-                            log::debug!("received request: {:?}", request);
-                            let resp = handle(service.as_ref(), request).await;
-                            log::debug!("response to send: {:?}", resp);
-                            let mut output = output.lock().await;
-                            log::debug!("sending");
-                            let r = SinkExt::send(&mut *output, resp).await;
-                            log::debug!("sending done: {:?}", r);
-                        });
+                        log::debug!("received request: {:?}", request);
+                        let resp = handle(service.as_ref(), request).await;
+                        log::debug!("response to send: {:?}", resp);
+                        let mut output = output.lock().await;
+                        log::debug!("sending");
+                        let r = SinkExt::send(&mut *output, resp).await;
+                        log::debug!("sending done: {:?}", r);
                     }
                     Err(e) => {
                         log::error!("fail: {}", e);

diff --git a/service-bus/bus/src/connection.rs b/service-bus/bus/src/connection.rs
@@ -373,7 +373,7 @@ where
                     Err(Error::GsbFailure(String::from_utf8(chunk.into_bytes())?))
                 }
             };
-            let _ = ctx.spawn(
+            let _ = ctx.wait(
                 async move {
                     let s = r.send(item);
                     s.await