fix: OnEvent is not propagating data in some scenarios with postgres (getdozer#637)

* chore: simplify stream_pipeline_request

* fix: multiple client subscribe

* chore: add log if broadcast error

* fix: Fix `Field` to `grpc::types::Value` conversion

* chore: fix issue delay async

Co-authored-by: chubei <[email protected]>

Author: duonganhthu43

dozer-api/src/grpc/internal/internal_pipeline_server.rs

@@ -6,45 +6,90 @@ use crate::grpc::internal_grpc::{
 use crossbeam::channel::Receiver;
 use dozer_types::{crossbeam, log::info, models::app_config::Config, tracing::warn};
 use std::{net::ToSocketAddrs, pin::Pin};
-use tokio::runtime::Runtime;
-use tokio_stream::{wrappers::ReceiverStream, StreamExt};
+use tokio::{
+    runtime::Runtime,
+    sync::broadcast::{self, Sender},
+};
+use tokio_stream::wrappers::ReceiverStream;
 use tonic::{codegen::futures_core::Stream, transport::Server, Response, Status};
 
 pub struct InternalPipelineServer {
     app_config: Config,
-    receiver: Receiver<PipelineResponse>,
+    receiver: broadcast::Receiver<PipelineResponse>,
+}
+impl InternalPipelineServer {
+    pub fn new(app_config: Config, receiver: Receiver<PipelineResponse>) -> Self {
+        let (tx, rx1) = broadcast::channel::<PipelineResponse>(16);
+        tokio::spawn(async move {
+            Self::setup_broad_cast_channel(tx, receiver);
+        });
+        Self {
+            app_config,
+            receiver: rx1,
+        }
+    }
+
+    fn setup_broad_cast_channel(
+        tx: Sender<PipelineResponse>,
+        receiver: Receiver<PipelineResponse>,
+    ) {
+        loop {
+            let message = receiver.recv();
+            match message {
+                Ok(message) => {
+                    let result = tx.send(message);
+                    if let Err(e) = result {
+                        warn!("Internal Pipeline server - Error sending message to broadcast channel: {:?}", e);
+                    }
+                }
+                Err(err) => {
+                    warn!(
+                        "Internal Pipeline server - message reveived error: {:?}",
+                        err
+                    );
+                    break;
+                }
+            }
+        }
+    }
 }
 type ResponseStream = Pin<Box<dyn Stream<Item = Result<PipelineResponse, Status>> + Send>>;
 
 #[tonic::async_trait]
 impl InternalPipelineService for InternalPipelineServer {
     type StreamPipelineRequestStream = ResponseStream;
-
     async fn stream_pipeline_request(
         &self,
         _request: tonic::Request<PipelineRequest>,
     ) -> Result<Response<ResponseStream>, Status> {
         let (tx, rx) = tokio::sync::mpsc::channel(1000);
-        let iterator = InternalIterator {
-            receiver: self.receiver.to_owned(),
-        };
-        let in_stream = tokio_stream::iter(iterator);
-        let mut stream = Box::pin(in_stream);
+        let mut receiver = self.receiver.resubscribe();
         tokio::spawn(async move {
-            while let Some(item) = stream.next().await {
-                if let Err(_item) = tx.send(Result::<_, Status>::Ok(item)).await {
-                    warn!("output_stream was build from rx and both are dropped");
-                    break;
+            loop {
+                let result = receiver.try_recv();
+                match result {
+                    Ok(message) => {
+                        let result = tx.send(Ok(message)).await;
+                        if let Err(e) = result {
+                            warn!("Error sending message to mpsc channel: {:?}", e);
+                            break;
+                        }
+                    }
+                    Err(err) => {
+                        if err == broadcast::error::TryRecvError::Closed {
+                            break;
+                        }
+                    }
                 }
+                tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
             }
-            warn!("client disconnected");
         });
-
         let output_stream = ReceiverStream::new(rx);
         Ok(Response::new(
             Box::pin(output_stream) as Self::StreamPipelineRequestStream
         ))
     }
+
     async fn get_config(
         &self,
         _request: tonic::Request<GetAppConfigRequest>,
@@ -72,10 +117,7 @@ async fn _start_internal_pipeline_server(
     app_config: Config,
     receiver: Receiver<PipelineResponse>,
 ) -> Result<(), tonic::transport::Error> {
-    let server = InternalPipelineServer {
-        app_config: app_config.to_owned(),
-        receiver,
-    };
+    let server = InternalPipelineServer::new(app_config.to_owned(), receiver);
 
     let internal_config = app_config
         .api
@@ -95,17 +137,3 @@ async fn _start_internal_pipeline_server(
         .serve(addr.next().unwrap())
         .await
 }
-
-struct InternalIterator {
-    receiver: Receiver<PipelineResponse>,
-}
-impl Iterator for InternalIterator {
-    type Item = PipelineResponse;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        match self.receiver.recv() {
-            Ok(msg) => Some(msg),
-            Err(_) => None,
-        }
-    }
-}

dozer-api/src/grpc/types_helper.rs

@@ -47,7 +47,7 @@ fn field_to_prost_value(f: Field) -> Value {
             value: Some(value::Value::IntValue(n)),
         },
         Field::Float(n) => Value {
-            value: Some(value::Value::FloatValue(n.0 as f32)),
+            value: Some(value::Value::DoubleValue(n.0)),
         },
 
         Field::Boolean(n) => Value {