logos-co/nomos-node
1
0
Fork 0
mirror of synced 2025-03-02 16:30:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Sync bound to BoxStream in blend stream (#908)

Browse Source
This commit is contained in:
Youngjoon Lee 2024-11-06 18:59:06 +09:00 committed by GitHub
parent 9b29c17e2f
commit e0959644a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
nomos-mix/core/src/message_blend/mod.rs
View File

@ -2,7 +2,6 @@ pub mod crypto;
pub mod temporal; pub mod temporal;
pub use crypto::CryptographicProcessorSettings; pub use crypto::CryptographicProcessorSettings;
use futures::stream::BoxStream;
use futures::{Stream, StreamExt}; use futures::{Stream, StreamExt};
use rand::RngCore; use rand::RngCore;
use std::marker::PhantomData; use std::marker::PhantomData;
@ -39,7 +38,7 @@ where
M: MixMessage, M: MixMessage,
{ {
input_stream: S, input_stream: S,
output_stream: BoxStream<'static, MixOutgoingMessage>, output_stream: Pin<Box<dyn Stream<Item = MixOutgoingMessage> + Send + Sync + 'static>>,
temporal_sender: UnboundedSender<MixOutgoingMessage>, temporal_sender: UnboundedSender<MixOutgoingMessage>,
cryptographic_processor: CryptographicProcessor<Rng, M>, cryptographic_processor: CryptographicProcessor<Rng, M>,
_rng: PhantomData<Rng>, _rng: PhantomData<Rng>,
@ -53,7 +52,7 @@ where
M: MixMessage, M: MixMessage,
M::PrivateKey: Serialize + DeserializeOwned, M::PrivateKey: Serialize + DeserializeOwned,
M::PublicKey: Clone + PartialEq, M::PublicKey: Clone + PartialEq,
Scheduler: Stream<Item = ()> + Unpin + Send + 'static, Scheduler: Stream<Item = ()> + Unpin + Send + Sync + 'static,
{ {
pub fn new( pub fn new(
input_stream: S, input_stream: S,
@ -68,9 +67,8 @@ where
cryptographic_processor_rng, cryptographic_processor_rng,
); );
let (temporal_sender, temporal_receiver) = mpsc::unbounded_channel(); let (temporal_sender, temporal_receiver) = mpsc::unbounded_channel();
let output_stream = UnboundedReceiverStream::new(temporal_receiver) let output_stream =
.temporal_stream(scheduler) Box::pin(UnboundedReceiverStream::new(temporal_receiver).temporal_stream(scheduler));
.boxed();
Self { Self {
input_stream, input_stream,
output_stream, output_stream,
@ -110,7 +108,7 @@ where
M: MixMessage + Unpin, M: MixMessage + Unpin,
M::PrivateKey: Serialize + DeserializeOwned + Unpin, M::PrivateKey: Serialize + DeserializeOwned + Unpin,
M::PublicKey: Clone + PartialEq + Unpin, M::PublicKey: Clone + PartialEq + Unpin,
Scheduler: Stream<Item = ()> + Unpin + Send + 'static, Scheduler: Stream<Item = ()> + Unpin + Send + Sync + 'static,
{ {
type Item = MixOutgoingMessage; type Item = MixOutgoingMessage;
@ -128,7 +126,7 @@ where
M: MixMessage, M: MixMessage,
M::PrivateKey: Serialize + DeserializeOwned, M::PrivateKey: Serialize + DeserializeOwned,
M::PublicKey: Clone + PartialEq, M::PublicKey: Clone + PartialEq,
Scheduler: Stream<Item = ()> + Unpin + Send + 'static, Scheduler: Stream<Item = ()> + Unpin + Send + Sync + 'static,
{ {
fn blend( fn blend(
self, self,
@ -157,6 +155,6 @@ where
M: MixMessage, M: MixMessage,
M::PrivateKey: Clone + Serialize + DeserializeOwned + PartialEq, M::PrivateKey: Clone + Serialize + DeserializeOwned + PartialEq,
M::PublicKey: Clone + Serialize + DeserializeOwned + PartialEq, M::PublicKey: Clone + Serialize + DeserializeOwned + PartialEq,
S: Stream<Item = ()> + Unpin + Send + 'static, S: Stream<Item = ()> + Unpin + Send + Sync + 'static,
{ {
} }