-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
17 changed files
with
505 additions
and
145 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| fn main() -> Result<(), Box<dyn std::error::Error>> { | ||
| tonic_build::compile_protos("proto/delegator.proto")?; | ||
| Ok(()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| syntax = "proto3"; | ||
|
|
||
| package delegator; | ||
|
|
||
| service DelegatorService { | ||
| rpc delegate(DelegateRequest) returns (DelegateResponse) {} | ||
| } | ||
|
|
||
| message DelegateRequest { string name = 1; } | ||
| message DelegateResponse { string message = 1; } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| pub mod event_loop; | ||
|
|
||
| use event_loop::delegator_loop; | ||
| use futures::executor::block_on; | ||
| use futures::FutureExt; | ||
| use libp2p::gossipsub::Event; | ||
| use starknet::signers::SigningKey; | ||
| use std::hash::{DefaultHasher, Hash, Hasher}; | ||
| use tokio::{sync::mpsc, task::JoinHandle}; | ||
| use tokio_util::sync::CancellationToken; | ||
| use tracing::info; | ||
| use zetina_common::{ | ||
| hash, | ||
| job::{Job, JobData}, | ||
| job_witness::JobWitness, | ||
| }; | ||
|
|
||
| pub struct Delegator<'identity> { | ||
| signing_key: &'identity SigningKey, | ||
| job_topic_tx: mpsc::Sender<Vec<u8>>, | ||
| cancellation_token: CancellationToken, | ||
| handle: Option<JoinHandle<()>>, | ||
| } | ||
|
|
||
| impl<'identity> Delegator<'identity> { | ||
| pub fn new( | ||
| signing_key: &'identity SigningKey, | ||
| job_topic_tx: mpsc::Sender<Vec<u8>>, | ||
| job_witness_tx: mpsc::Sender<JobWitness>, | ||
| events_rx: mpsc::Receiver<Event>, | ||
| ) -> Self { | ||
| let cancellation_token = CancellationToken::new(); | ||
|
|
||
| Self { | ||
| signing_key, | ||
| job_topic_tx, | ||
| cancellation_token: cancellation_token.to_owned(), | ||
| handle: Some(tokio::spawn(async move { | ||
| delegator_loop(events_rx, job_witness_tx, cancellation_token).boxed().await | ||
| })), | ||
| } | ||
| } | ||
|
|
||
| pub async fn delegate(self, job_data: JobData) -> Result<(), mpsc::error::SendError<Vec<u8>>> { | ||
| let job = Job::try_from_job_data(job_data, self.signing_key); | ||
| let serialized_job = serde_json::to_string(&job).unwrap(); | ||
| self.job_topic_tx.send(serialized_job.into()).await?; | ||
| info!("Sent a new job: {}", hash!(&job)); | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| impl<'identity> Drop for Delegator<'identity> { | ||
| fn drop(&mut self) { | ||
| self.cancellation_token.cancel(); | ||
| block_on(async move { | ||
| self.handle.take().unwrap().await.unwrap(); | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| use libp2p::gossipsub::Event; | ||
| use std::hash::{DefaultHasher, Hash, Hasher}; | ||
| use tokio::sync::mpsc; | ||
| use tokio_util::sync::CancellationToken; | ||
| use tracing::info; | ||
| use zetina_common::{ | ||
| hash, | ||
| job::Job, | ||
| job_witness::JobWitness, | ||
| network::Network, | ||
| topic::{gossipsub_ident_topic, Topic}, | ||
| }; | ||
|
|
||
| pub async fn delegator_loop( | ||
| mut message_stream: mpsc::Receiver<Event>, | ||
| job_witness_tx: mpsc::Sender<JobWitness>, | ||
| cancellation_token: CancellationToken, | ||
| ) { | ||
| loop { | ||
| tokio::select! { | ||
| Some(event) = message_stream.recv() => { | ||
| match event { | ||
| Event::Message { message, .. } => { | ||
| // Received a new-job message from the network | ||
| if message.topic == gossipsub_ident_topic(Network::Sepolia, Topic::NewJob).into() { | ||
| let job: Job = serde_json::from_slice(&message.data).unwrap(); | ||
| info!("Received a new job event: {}", hash!(&job)); | ||
| } | ||
| // Received a picked-job message from the network | ||
| if message.topic == gossipsub_ident_topic(Network::Sepolia, Topic::PickedJob).into() { | ||
| let job: Job = serde_json::from_slice(&message.data).unwrap(); | ||
| info!("Received picked job event: {}", hash!(&job)); | ||
| } | ||
| // Received a finished-job message from the network | ||
| if message.topic == gossipsub_ident_topic(Network::Sepolia, Topic::FinishedJob).into() { | ||
| let job_witness: JobWitness = serde_json::from_slice(&message.data).unwrap(); | ||
| info!("Received finished job event: {}", hash!(&job_witness)); | ||
| job_witness_tx.send(job_witness).await.unwrap(); | ||
| } | ||
| }, | ||
| Event::Subscribed { peer_id, topic } => { | ||
| info!("{} subscribed to the topic {}", peer_id.to_string(), topic.to_string()); | ||
| }, | ||
| Event::Unsubscribed { peer_id, topic }=> { | ||
| info!("{} unsubscribed to the topic {}", peer_id.to_string(), topic.to_string()); | ||
| }, | ||
| _ => {} | ||
| } | ||
| }, | ||
| _ = cancellation_token.cancelled() => { | ||
| break | ||
| } | ||
| else => break | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.