From 6980636686bb2a26b7731cd766e31b955e88a7ed Mon Sep 17 00:00:00 2001
From: Philipp Oppermann <dev@phil-opp.com>
Date: Wed, 8 Feb 2023 21:26:29 +0100
Subject: [PATCH] Update futures-concurrency to v7.1

---
 Cargo.lock                      | 31 +++++--------------------------
 binaries/coordinator/Cargo.toml |  2 +-
 binaries/daemon/Cargo.toml      |  2 +-
 binaries/runtime/Cargo.toml     |  2 +-
 binaries/runtime/src/lib.rs     |  6 +++---
 5 files changed, 11 insertions(+), 32 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 6f8d0ca3..7184e3fc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -962,7 +962,7 @@ dependencies = [
  "dora-node-api",
  "eyre",
  "futures",
- "futures-concurrency 5.0.1",
+ "futures-concurrency",
  "rand",
  "serde",
  "serde_json",
@@ -1005,7 +1005,7 @@ dependencies = [
  "eyre",
  "flume",
  "futures",
- "futures-concurrency 7.0.0",
+ "futures-concurrency",
  "serde",
  "serde_json",
  "serde_yaml 0.8.23",
@@ -1192,7 +1192,7 @@ dependencies = [
  "fern",
  "flume",
  "futures",
- "futures-concurrency 2.0.3",
+ "futures-concurrency",
  "libloading",
  "opentelemetry",
  "opentelemetry-system-metrics",
@@ -1388,30 +1388,9 @@ dependencies = [
 
 [[package]]
 name = "futures-concurrency"
-version = "2.0.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48e98b7b5aedee7c34a5cfb1ee1681af8faf46e2f30c0b8af5ea08eba517d61c"
-dependencies = [
- "async-trait",
- "futures-core",
- "pin-project",
-]
-
-[[package]]
-name = "futures-concurrency"
-version = "5.0.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "407ed2aa475d777e35fb167144b63babd0377b2f9a528ae3ec4bec94f1ce1f1a"
-dependencies = [
- "futures-core",
- "pin-project",
-]
-
-[[package]]
-name = "futures-concurrency"
-version = "7.0.0"
+version = "7.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a740c32e1bde284ce2f51df98abd4fa38e9e539670443c111211777e3ab09927"
+checksum = "e06f199437c8a435c12ad153c5a1f4e131871cf6f6025585bb15e8cbb414f0dc"
 dependencies = [
  "bitvec",
  "futures-core",
diff --git a/binaries/coordinator/Cargo.toml b/binaries/coordinator/Cargo.toml
index 0bec9dd6..1a75c20f 100644
--- a/binaries/coordinator/Cargo.toml
+++ b/binaries/coordinator/Cargo.toml
@@ -24,7 +24,7 @@ dora-core = { workspace = true }
 dora-message = { path = "../../libraries/message" }
 tracing = "0.1.36"
 tracing-subscriber = "0.3.15"
-futures-concurrency = "5.0.1"
+futures-concurrency = "7.1.0"
 zenoh = { git = "https://github.com/eclipse-zenoh/zenoh.git", rev = "79a136e4fd90b11ff5d775ced981af53c4f1071b" }
 serde_json = "1.0.86"
 dora-download = { path = "../../libraries/extensions/download" }
diff --git a/binaries/daemon/Cargo.toml b/binaries/daemon/Cargo.toml
index 6be78b6a..3fcf4f13 100644
--- a/binaries/daemon/Cargo.toml
+++ b/binaries/daemon/Cargo.toml
@@ -11,7 +11,7 @@ tokio = { version = "1.20.1", features = ["full"] }
 tokio-stream = { version = "0.1.11", features = ["net"] }
 tracing = "0.1.36"
 tracing-subscriber = "0.3.15"
-futures-concurrency = "7.0.0"
+futures-concurrency = "7.1.0"
 serde = { version = "1.0.136", features = ["derive"] }
 serde_json = "1.0.86"
 dora-core = { path = "../../libraries/core" }
diff --git a/binaries/runtime/Cargo.toml b/binaries/runtime/Cargo.toml
index 033db80c..c94bb672 100644
--- a/binaries/runtime/Cargo.toml
+++ b/binaries/runtime/Cargo.toml
@@ -21,7 +21,7 @@ opentelemetry = { version = "0.17", features = [
 opentelemetry-system-metrics = { version = "0.1.1", optional = true }
 eyre = "0.6.8"
 futures = "0.3.21"
-futures-concurrency = "2.0.3"
+futures-concurrency = "7.1.0"
 libloading = "0.7.3"
 serde_yaml = "0.8.23"
 tokio = { version = "1.17.0", features = ["full"] }
diff --git a/binaries/runtime/src/lib.rs b/binaries/runtime/src/lib.rs
index f1336648..829b2710 100644
--- a/binaries/runtime/src/lib.rs
+++ b/binaries/runtime/src/lib.rs
@@ -8,7 +8,7 @@ use dora_core::{
 use dora_node_api::DoraNode;
 use eyre::{bail, Context, Result};
 use futures::{Stream, StreamExt};
-use futures_concurrency::Merge;
+use futures_concurrency::stream::Merge;
 use operator::{run_operator, OperatorEvent, StopReason};
 
 use std::{collections::HashMap, mem};
@@ -47,7 +47,7 @@ pub fn main() -> eyre::Result<()> {
         id: operator_id.clone(),
         event,
     });
-    let daemon_events = futures::stream::unfold(daemon_events, |mut stream| async {
+    let daemon_events = Box::pin(futures::stream::unfold(daemon_events, |mut stream| async {
         let event = stream.recv_async().await.map(|event| match event {
             dora_node_api::daemon::Event::Stop => Event::Stop,
             dora_node_api::daemon::Event::Input { id, metadata, data } => Event::Input {
@@ -60,7 +60,7 @@ pub fn main() -> eyre::Result<()> {
             _ => todo!(),
         });
         event.map(|event| (event, stream))
-    });
+    }));
     let events = (operator_events, daemon_events).merge();
     let tokio_runtime = Builder::new_current_thread()
         .enable_all()