Order pages now update in realtime

napoli-lib/proto/comms.proto

@@ -57,4 +57,7 @@ service OrderService {
     rpc AddOrderEntry (AddOrderEntryRequest) returns (SingleOrderReply);
     rpc RemoveOrderEntry (OrderEntryRequest) returns (SingleOrderReply);
     rpc SetOrderEntryPaid (SetOrderEntryPaidRequest) returns (SingleOrderReply);
+
+    // Live Updates
+    rpc StreamOrderUpdates (GetOrderRequest) returns (stream SingleOrderReply);
 }

napoli-pain/Cargo.toml

@@ -20,3 +20,4 @@ prost = "0"
 tonic-web-wasm-client = "0"
 tonic = { version = "0.8.3", default-features = false, features = ["codegen", "prost"] }
 web-sys = "0"
+futures = "0.3.28"

napoli-pain/src/components/order_details/live_streaming_indicator.rs

@@ -0,0 +1,28 @@
+#[derive(PartialEq, Clone)]
+pub enum LiveStreamingStatus {
+    Error(String),
+    Connecting,
+    Connected,
+}
+
+#[derive(yew::Properties, PartialEq, Clone)]
+pub struct StreamingIndicatorProps {
+    pub status: LiveStreamingStatus,
+}
+
+#[yew::function_component(StreamingIndicator)]
+pub fn streaming_indicator(props: &StreamingIndicatorProps) -> yew::Html {
+    let (status_symbol, status_message, hover_text) = match &props.status {
+        LiveStreamingStatus::Error(msg) => {
+            ("🔴", "Live updates inactive", format!("Error: {}", msg))
+        }
+        LiveStreamingStatus::Connecting => ("🟡", "Live updates connecting", String::new()),
+        LiveStreamingStatus::Connected => ("🟢", "Live updates active", String::new()),
+    };
+
+    yew::html!(
+        <div class="fixed top-4 right-4" title={hover_text}>
+            {status_message}{" "}{status_symbol}
+        </div>
+    )
+}

napoli-pain/src/components/order_details/mod.rs

@@ -1,2 +1,3 @@
 mod add_order_entry_form;
+mod live_streaming_indicator;
 pub mod order_details;

napoli-pain/src/components/order_details/order_details.rs

@@ -1,10 +1,13 @@
 use crate::{
     components::order_details::add_order_entry_form::AddOrderEntryForm,
+    components::order_details::live_streaming_indicator::{
+        LiveStreamingStatus, StreamingIndicator,
+    },
     router::Route,
     service::{self},
 };
-
-use napoli_lib::napoli::{self as npb, ObjectId};
+use futures::StreamExt;
+use napoli_lib::napoli::{self as npb, ObjectId, SingleOrderReply};
 use yew::prelude::*;
 use yew_router::prelude::Link;
 
@@ -15,6 +18,7 @@ pub struct OrderDetailsProps {
 
 pub struct OrderDetails {
     order: Option<npb::Order>,
+    live_streaming_status: LiveStreamingStatus,
 }
 
 pub enum OrderDetailsMsg {
@@ -24,6 +28,10 @@ pub enum OrderDetailsMsg {
     AddOrderEntry(npb::AddOrderEntryRequest),
     SetOrderEntryPaid { entry_id: ObjectId, paid: bool },
     RemoveOrderEntry { entry_id: ObjectId },
+
+    StreamingConnected(tonic::Streaming<npb::SingleOrderReply>),
+    GotStreamingOrderUpdate(npb::Order),
+    StreamingFailed(service::ServiceError),
 }
 
 impl Component for OrderDetails {
@@ -32,13 +40,28 @@ impl Component for OrderDetails {
 
     fn create(ctx: &Context<Self>) -> Self {
         let mut svc = service::Napoli::new(crate::BACKEND_URL.to_string());
+
         ctx.link().send_future(async move {
             match svc.get_orders().await {
                 Ok(orders) => Self::Message::GotOrders(orders),
                 Err(e) => Self::Message::OrderFetchFailed(e),
             }
         });
-        Self { order: None }
+
+        let mut svc = service::Napoli::new(crate::BACKEND_URL.to_string());
+        let id = ctx.props().id;
+        ctx.link().send_future(async move {
+            let res = svc.stream_order_updates(id).await;
+            match res {
+                Ok(stream) => Self::Message::StreamingConnected(stream),
+                Err(e) => Self::Message::StreamingFailed(e),
+            }
+        });
+
+        Self {
+            order: None,
+            live_streaming_status: LiveStreamingStatus::Connecting,
+        }
     }
 
     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
@@ -47,6 +70,10 @@ impl Component for OrderDetails {
                 self.order = o.into_iter().find(|order| order.id == ctx.props().id);
                 true
             }
+            Self::Message::GotStreamingOrderUpdate(o) => {
+                self.order = Some(o);
+                true
+            }
             Self::Message::OrderFetchFailed(_e) => false,
             Self::Message::SetOrderEntryPaid { entry_id, paid } => {
                 let mut svc = service::Napoli::new(crate::BACKEND_URL.to_string());
@@ -90,6 +117,30 @@ impl Component for OrderDetails {
                 self.order = Some(order);
                 true
             }
+            Self::Message::StreamingConnected(stream) => {
+                self.live_streaming_status = LiveStreamingStatus::Connected;
+                ctx.link()
+                    .send_stream(stream.map(|single_order_reply_result| {
+                        match single_order_reply_result {
+                            Ok(SingleOrderReply { order: Some(o) }) => {
+                                return Self::Message::GotStreamingOrderUpdate(o);
+                            }
+                            Ok(SingleOrderReply { order: None }) => Self::Message::StreamingFailed(
+                                service::ServiceError::from("Got empty order"),
+                            ),
+                            Err(e) => {
+                                return Self::Message::StreamingFailed(
+                                    service::ServiceError::from(e),
+                                );
+                            }
+                        }
+                    }));
+                true
+            }
+            Self::Message::StreamingFailed(e) => {
+                self.live_streaming_status = LiveStreamingStatus::Error(format!("{:?}", e));
+                true
+            }
         }
     }
 
@@ -136,6 +187,7 @@ impl Component for OrderDetails {
                     </ul>
                     <AddOrderEntryForm order_id={order.id} onclick={on_add_new_order_request} />
                     <OrderSummary order_entries={order.entries.clone()} />
+                    <StreamingIndicator status={self.live_streaming_status.clone()} />
                 </div>
             }
         } else {

napoli-pain/src/service.rs

@@ -1,8 +1,7 @@
 use tonic_web_wasm_client::Client;
 use yew::prelude::*;
 
-use napoli_lib::napoli::order_service_client as npb_grpc;
-use napoli_lib::napoli::{self as npb, ObjectId};
+use napoli_lib::napoli as npb;
 
 #[derive(Debug, Clone)]
 pub struct ServiceError(String);
@@ -21,16 +20,22 @@ impl From<tonic::Status> for ServiceError {
     }
 }
 
+impl From<&str> for ServiceError {
+    fn from(other: &str) -> Self {
+        ServiceError(other.into())
+    }
+}
+
 type Result<T> = std::result::Result<T, ServiceError>;
 
 pub struct Napoli {
-    pub client: npb_grpc::OrderServiceClient<Client>,
+    pub client: npb::order_service_client::OrderServiceClient<Client>,
 }
 
 impl Napoli {
     pub fn new(backend_url: String) -> Self {
         Napoli {
-            client: npb_grpc::OrderServiceClient::new(Client::new(backend_url)),
+            client: npb::order_service_client::OrderServiceClient::new(Client::new(backend_url)),
         }
     }
 
@@ -41,8 +46,8 @@ impl Napoli {
 
     pub async fn set_order_entry_paid(
         &mut self,
-        order_id: ObjectId,
-        order_entry_id: ObjectId,
+        order_id: npb::ObjectId,
+        order_entry_id: npb::ObjectId,
         paid: bool,
     ) -> Result<npb::Order> {
         let order = self
@@ -79,4 +84,13 @@ impl Napoli {
         let order = self.client.remove_order_entry(request).await?;
         Ok(order.into_inner().order.expect("fucked up"))
     }
+
+    pub async fn stream_order_updates(
+        &mut self,
+        order_id: i32,
+    ) -> Result<tonic::Streaming<npb::SingleOrderReply>> {
+        let request = npb::GetOrderRequest { order_id };
+        let res = self.client.stream_order_updates(request).await?;
+        Ok(res.into_inner())
+    }
 }

napoli-server/Cargo.toml

@@ -19,6 +19,7 @@ clap = { version = "4.1.4", features = ["derive"] }
 tonic-web = "0.5.0"
 tower-http = "0"
 http = "0"
+tokio-stream = { version = "0.1.14", features = ["sync"] }
 
 # client binary
 [[bin]]

napoli-server/src/main.rs

@@ -40,6 +40,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     println!("NapoliServer listening on {}", addr);
     let napoli_server = NapoliServer::with_connection(db);
+
     let order_service_server = OrderServiceServer::new(napoli_server);
     let reflection = tonic_reflection::server::Builder::configure()
         .register_encoded_file_descriptor_set(FILE_DESCRIPTOR_SET)