lores_app_node/stores/
grpc.rs

1use std::pin::Pin;
2use std::sync::Arc;
3
4use futures::StreamExt;
5use lores_p2panda_client::{PandaClient, PandaError, PublishResult};
6use tokio::sync::Mutex;
7
8use crate::{
9    stores::{OperationStore, OperationStream, RawOperationEvent, StoreError, StorePublishResult},
10    NodeId, OperationId,
11};
12
13impl From<PandaError> for StoreError {
14    fn from(e: PandaError) -> Self {
15        match e {
16            PandaError::RegionNotBound(msg) => StoreError::RegionNotBound(msg),
17            PandaError::Rpc(s) => StoreError::Other(s.to_string()),
18        }
19    }
20}
21
22/// [`OperationStore`] implementation that forwards operations to a lores-node
23/// instance via gRPC using [`PandaClient`].
24pub(crate) struct GrpcOperationStore {
25    client: Arc<Mutex<PandaClient>>,
26    app_id: String,
27    instance_id: String,
28}
29
30impl GrpcOperationStore {
31    pub(crate) fn new(client: Arc<Mutex<PandaClient>>, app_id: impl Into<String>, instance_id: impl Into<String>) -> Self {
32        Self {
33            client,
34            app_id: app_id.into(),
35            instance_id: instance_id.into(),
36        }
37    }
38}
39
40impl OperationStore for GrpcOperationStore {
41    fn publish(
42        &mut self,
43        payload: Vec<u8>,
44        idempotency_key: Option<String>,
45    ) -> Pin<Box<dyn std::future::Future<Output = Result<StorePublishResult, StoreError>> + Send + '_>> {
46        Box::pin(async move {
47            let PublishResult { operation_id, node_id } = self
48                .client
49                .lock()
50                .await
51                .publish(&self.app_id, &self.instance_id, payload, idempotency_key.map(|k| k.into_bytes()))
52                .await
53                .map_err(StoreError::from)?;
54            Ok(StorePublishResult {
55                operation_id: operation_id.into_non_empty().map(OperationId),
56                node_id: node_id.into_non_empty().map(NodeId),
57            })
58        })
59    }
60
61    fn subscribe(&mut self) -> Pin<Box<dyn std::future::Future<Output = Result<OperationStream, StoreError>> + Send + '_>> {
62        Box::pin(async move {
63            let response = self
64                .client
65                .lock()
66                .await
67                .subscribe(&self.app_id, &self.instance_id)
68                .await
69                .map_err(StoreError::from)?;
70
71            let stream: OperationStream = Box::pin(response.into_inner().map(|item| {
72                item.map(|event| RawOperationEvent {
73                    payload: event.payload,
74                    author: Some(event.author),
75                    operation_id: Some(event.operation_id),
76                    timestamp: Some(event.timestamp),
77                })
78                .map_err(|s| StoreError::Other(s.to_string()))
79            }));
80
81            Ok(stream)
82        })
83    }
84}
lores_app_node/stores/grpc.rs

lores_app_node/stores/
grpc.rs
