pub async fn new(mesh: FullMesh<T>, shard_fn: ShardFn<T>, handler: H) -> Result<Self, ()> {
        let nr_shards = mesh.nr_peers();

        let (senders, mut receivers) = mesh.join().await?;

        let shard = Rc::new(Shard {
            nr_shards,
            shard_id: senders.peer_id(),
            shard_fn,
            senders,
            handler: handler.clone(),
        });

        let mut forward_tasks = Vec::with_capacity(nr_shards);
        for (src_shard, stream) in receivers.streams() {
            let handler = handler.clone();
            let cur_shard = shard.shard_id;
            let consumer = crate::spawn_local(async move {
                while let Some(msg) = stream.recv().await {
                    handler.handle(msg, src_shard, cur_shard).await;
                }
            });
            forward_tasks.push(consumer.detach());
        }

        Ok(Self {
            shard,
            consumers: Vec::new(),
            forward_tasks,
            closed: false,
        })
    }
/// Join a full mesh for sharding
Join a full mesh for sharding
Join a full mesh for sharding
pub async fn new(mesh: FullMesh<T>, shard_fn: ShardFn<T>, handler: H) -> Result<Self, ()> {
        let nr_shards = mesh.nr_peers();
        let (senders, mut receivers) = mesh.join().await?;
        let shard = Rc::new(Shard {
            nr_shards,
            shard_id: senders.peer_id(),
            shard_fn,
            senders,
            handler: handler.clone(),
        });
        let mut forward_tasks = Vec::with_capacity(nr_shards);
        for (src_shard, stream) in receivers.streams() {
            let handler = handler.clone();
            let cur_shard = shard.shard_id;
            let consumer = crate::spawn_local(async move {
                while let Some(msg) = stream.recv().await {
                    handler.handle(msg, src_shard, cur_shard).await;
                }
            });
            forward_tasks.push(consumer.detach());
        }
        Ok(Self {
            shard,
            consumers: Vec::new(),
            forward_tasks,
            closed: false,
        })
    }
pub async fn new ( mesh : FullMesh < T > , shard_fn : ShardFn < T > , handler : H ) -> Result < Self , ( ) > { let nr_shards = mesh . nr_peers ( ) ; let ( senders , mut receivers ) = mesh . join ( ) . await ? ; let shard = Rc :: new ( Shard { nr_shards , shard_id : senders . peer_id ( ) , shard_fn , senders , handler : handler . clone ( ) , } ) ; let mut forward_tasks = Vec :: with_capacity ( nr_shards ) ; for ( src_shard , stream ) in receivers . streams ( ) { let handler = handler . clone ( ) ; let cur_shard = shard . shard_id ; let consumer = crate :: spawn_local ( async move { while let Some ( msg ) = stream . recv ( ) . await { handler . handle ( msg , src_shard , cur_shard ) . await ; } } ) ; forward_tasks . push ( consumer . detach ( ) ) ; } Ok ( Self { shard , consumers : Vec :: new ( ) , forward_tasks , closed : false , } ) }
Join a full mesh for sharding
Join a full mesh for sharding
mesh
FullMesh<T>
shard_fn
ShardFn<T>
handler
H
mesh
FullMesh<T>
shard_fn
ShardFn<T>
handler
H