-
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
4 changed files
with
73 additions
and
167 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,131 +1,53 @@ | ||
| // Declare the entity | ||
|
|
||
| use std::{io, num::NonZeroUsize, pin::Pin}; | ||
|
|
||
| use akka_persistence_rs::{ | ||
| entity_manager::{self, EventEnvelope, Handler, SourceProvider}, | ||
| EntityId, Message, | ||
| }; | ||
| use async_trait::async_trait; | ||
| use akka_persistence_rs::{entity_manager::EventEnvelope, EntityId}; | ||
| use gloo_net::eventsource::futures::EventSource; | ||
| use log::{error, warn}; | ||
| use tokio::sync::{broadcast, mpsc}; | ||
| use tokio_stream::{Stream, StreamExt}; | ||
| use yew::platform; | ||
| use tokio_stream::StreamExt; | ||
|
|
||
| pub use iot_service_model::temperature::{Behavior, Command, Event, SecretDataValue, State}; | ||
|
|
||
| // FIXME this adapter should be generalised | ||
|
|
||
| pub struct EventSourceAdapter { | ||
| query: mpsc::UnboundedSender<EntityId>, | ||
| event: broadcast::Sender<EventEnvelope<Event>>, | ||
| } | ||
|
|
||
| impl EventSourceAdapter { | ||
| pub fn new( | ||
| query: mpsc::UnboundedSender<EntityId>, | ||
| event: broadcast::Sender<EventEnvelope<Event>>, | ||
| ) -> Self { | ||
| Self { query, event } | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl SourceProvider<Event> for EventSourceAdapter { | ||
| async fn source_initial( | ||
| &mut self, | ||
| ) -> io::Result<Pin<Box<dyn Stream<Item = EventEnvelope<Event>> + Send + 'async_trait>>> { | ||
| Ok(Box::pin(tokio_stream::empty())) | ||
| } | ||
|
|
||
| async fn source( | ||
| &mut self, | ||
| entity_id: &EntityId, | ||
| ) -> io::Result<Pin<Box<dyn Stream<Item = EventEnvelope<Event>> + Send + 'async_trait>>> { | ||
| if self.query.send(entity_id.clone()).is_ok() { | ||
| Ok(Box::pin(tokio_stream::empty())) | ||
| } else { | ||
| Err(io::Error::new(io::ErrorKind::Other, "Cannot send query")) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl Handler<Event> for EventSourceAdapter { | ||
| async fn process( | ||
| &mut self, | ||
| envelope: EventEnvelope<Event>, | ||
| ) -> io::Result<EventEnvelope<Event>> { | ||
| self.event | ||
| .send(envelope.clone()) | ||
| .map(|_| envelope) | ||
| .map_err(|_context| io::Error::new(io::ErrorKind::Other, "Cannot broadcast")) | ||
| } | ||
| } | ||
|
|
||
| // Execute the entity manager for this entity | ||
|
|
||
| pub async fn task( | ||
| command: mpsc::Sender<Message<Command>>, | ||
| command_receiver: mpsc::Receiver<Message<Command>>, | ||
| event: broadcast::Sender<EventEnvelope<Event>>, | ||
| mut query_receiver: mpsc::Receiver<EntityId>, | ||
| events: broadcast::Sender<EventEnvelope<Event>>, | ||
| ) { | ||
| let (query, mut query_receiver) = mpsc::unbounded_channel(); | ||
|
|
||
| platform::spawn_local(async move { | ||
| if let Some(mut entity_id) = query_receiver.recv().await { | ||
| 'outer: loop { | ||
| let url: &str = &format!("/api/temperature/events/{entity_id}"); | ||
| let mut temperature_es = EventSource::new(url).unwrap(); | ||
| let mut temperature_events = temperature_es.subscribe("message").unwrap(); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| Some(Ok((_, message))) = temperature_events.next() => { | ||
| let data: Option<String> = message.data().as_string(); | ||
|
|
||
| if let Some(data) = data { | ||
| let update_command = match serde_json::from_str::<Event>(&data) { | ||
| Ok(event) => match event { | ||
| Event::Registered { secret } => Some(Command::Register { secret }), | ||
| Event::TemperatureRead { temperature } => { | ||
| Some(Command::Post { temperature }) | ||
| } | ||
| }, | ||
| Err(e) => { | ||
| error!("Failed to parse event: {}", e); | ||
| None | ||
| } | ||
| }; | ||
| if let Some(update_command) = update_command { | ||
| let _ = command | ||
| .send(Message::new(EntityId::from(""), update_command)) | ||
| .await; | ||
| if let Some(mut entity_id) = query_receiver.recv().await { | ||
| 'outer: loop { | ||
| let url: &str = &format!("/api/temperature/events/{entity_id}"); | ||
| let mut temperature_es = EventSource::new(url).unwrap(); | ||
| let mut temperature_events = temperature_es.subscribe("message").unwrap(); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| Some(Ok((_, message))) = temperature_events.next() => { | ||
| let data: Option<String> = message.data().as_string(); | ||
|
|
||
| if let Some(data) = data { | ||
| match serde_json::from_str::<EventEnvelope<Event>>(&data) { | ||
| Ok(envelope) => { | ||
| let _ = events.send(envelope); | ||
| } | ||
| } else { | ||
| warn!("Received event with no data"); | ||
| } | ||
|
|
||
| Err(e) => { | ||
| error!("Failed to parse event: {}", e); | ||
| } | ||
| }; | ||
| } else { | ||
| warn!("Received event with no data"); | ||
| } | ||
|
|
||
| Some(next_entity_id) = query_receiver.recv() => { | ||
| entity_id = next_entity_id; | ||
| break | ||
| } | ||
| } | ||
|
|
||
| else => break 'outer, | ||
| Some(next_entity_id) = query_receiver.recv() => { | ||
| entity_id = next_entity_id; | ||
| break | ||
| } | ||
|
|
||
| else => break 'outer, | ||
| } | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| entity_manager::run( | ||
| Behavior, | ||
| EventSourceAdapter::new(query, event), | ||
| command_receiver, | ||
| NonZeroUsize::new(10).unwrap(), | ||
| ) | ||
| .await | ||
| } | ||
| } |
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