camel_component_grpc/
consumer.rs

1use std::collections::HashMap as StdHashMap;
2use std::path::PathBuf;
3use std::sync::Arc;
4use std::sync::OnceLock;
5use std::sync::atomic::{AtomicU64, Ordering};
6
7use async_trait::async_trait;
8use base64::Engine;
9use bytes::BytesMut;
10use camel_api::store_principal_properties;
11use camel_api::{AuthorizationDecision, Body, CamelError, Exchange, Message, Value};
12use camel_component_api::{
13    ConcurrencyModel, Consumer, ConsumerContext, ConsumerStartupMode, ExchangeEnvelope,
14    SecurityContext,
15};
16use camel_proto_compiler::ProtoCache;
17use prost::Message as _;
18use prost_reflect::{DynamicMessage, MessageDescriptor};
19use tokio::sync::mpsc;
20use tonic::Status;
21use tracing::{debug, info};
22
23use crate::config::GrpcServerConfig;
24use crate::mode::GrpcMode;
25use crate::server::GrpcDispatchTable;
26use crate::server::GrpcServerRegistry;
27
28static PROTO_CACHE: OnceLock<ProtoCache> = OnceLock::new();
29
30fn proto_cache() -> &'static ProtoCache {
31    PROTO_CACHE.get_or_init(ProtoCache::new)
32}
33
34/// Resolve the gRPC mode (unary/streaming) for a given method without creating a consumer.
35pub fn resolve_grpc_mode(
36    proto_path: &PathBuf,
37    service_name: &str,
38    method_name: &str,
39) -> Result<GrpcMode, CamelError> {
40    let cache = proto_cache();
41    let pool = cache
42        .get_or_compile(proto_path, std::iter::empty::<&std::path::Path>())
43        .map_err(|e| CamelError::EndpointCreationFailed(format!("failed to compile proto: {e}")))?;
44
45    let svc = pool.get_service_by_name(service_name).ok_or_else(|| {
46        CamelError::EndpointCreationFailed(format!(
47            "service descriptor not found: {}",
48            service_name
49        ))
50    })?;
51
52    let method = svc
53        .methods()
54        .find(|m| m.name() == method_name)
55        .ok_or_else(|| {
56            CamelError::EndpointCreationFailed(format!(
57                "method descriptor not found: {}/{}",
58                service_name, method_name
59            ))
60        })?;
61
62    Ok(GrpcMode::from_method(&method))
63}
64
65const RESERVED_METADATA_KEYS: &[&str] = &[
66    "content-type",
67    "te",
68    "grpc-encoding",
69    "grpc-accept-encoding",
70    "grpc-status",
71    "grpc-message",
72    "grpc-status-details-bin",
73    "user-agent",
74];
75
76fn extract_metadata(metadata: &tonic::metadata::MetadataMap) -> Vec<(String, serde_json::Value)> {
77    let mut headers = Vec::new();
78    for key_and_value in metadata.iter() {
79        use tonic::metadata::KeyAndValueRef;
80        match key_and_value {
81            KeyAndValueRef::Ascii(key, value) => {
82                let key_str = key.as_str();
83                if RESERVED_METADATA_KEYS.contains(&key_str) {
84                    continue;
85                }
86                if let Ok(v) = value.to_str() {
87                    headers.push((
88                        key_str.to_string(),
89                        serde_json::Value::String(v.to_string()),
90                    ));
91                }
92            }
93            KeyAndValueRef::Binary(key, value) => {
94                let key_str = key.as_str();
95                if RESERVED_METADATA_KEYS.contains(&key_str) {
96                    continue;
97                }
98                let encoded = base64::engine::general_purpose::STANDARD.encode(value);
99                headers.push((format!("bin:{key_str}"), serde_json::Value::String(encoded)));
100            }
101        }
102    }
103    headers
104}
105
106pub(crate) enum GrpcStreamItem {
107    Message(Vec<u8>),
108    Error(tonic::Status),
109    Done,
110}
111
112pub(crate) enum GrpcReply {
113    Ok(Vec<u8>),
114    Err(tonic::Status),
115}
116
117pub(crate) enum GrpcRequestEnvelope {
118    Unary {
119        metadata: tonic::metadata::MetadataMap,
120        body: Vec<u8>,
121        reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
122        principal: Option<camel_api::security_policy::Principal>,
123    },
124    ServerStreaming {
125        metadata: tonic::metadata::MetadataMap,
126        body: Vec<u8>,
127        reply_tx: mpsc::Sender<GrpcStreamItem>,
128        principal: Option<camel_api::security_policy::Principal>,
129    },
130    ClientStreaming {
131        metadata: tonic::metadata::MetadataMap,
132        body_rx: mpsc::Receiver<Vec<u8>>,
133        reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
134        principal: Option<camel_api::security_policy::Principal>,
135    },
136    Bidi {
137        metadata: tonic::metadata::MetadataMap,
138        body_rx: mpsc::Receiver<Vec<u8>>,
139        reply_tx: mpsc::Sender<GrpcStreamItem>,
140        principal: Option<camel_api::security_policy::Principal>,
141    },
142}
143
144// ── Observer registry ──────────────────────────────────────────────────────
145
146static OBSERVER_REGISTRY: OnceLock<std::sync::Mutex<StdHashMap<String, GrpcStreamObserver>>> =
147    OnceLock::new();
148
149static OBSERVER_COUNTER: AtomicU64 = AtomicU64::new(0);
150
151fn next_observer_id() -> String {
152    let n = OBSERVER_COUNTER.fetch_add(1, Ordering::Relaxed);
153    format!("obs-{n}")
154}
155
156fn observer_registry() -> &'static std::sync::Mutex<StdHashMap<String, GrpcStreamObserver>> {
157    OBSERVER_REGISTRY.get_or_init(|| std::sync::Mutex::new(StdHashMap::new()))
158}
159
160fn register_observer(id: String, observer: GrpcStreamObserver) {
161    let registry = observer_registry();
162    let mut registry = match registry.lock() {
163        Ok(g) => g,
164        Err(poisoned) => poisoned.into_inner(),
165    };
166    registry.insert(id, observer);
167}
168
169fn remove_observer(id: &str) -> Option<GrpcStreamObserver> {
170    let registry = observer_registry();
171    let mut registry = match registry.lock() {
172        Ok(g) => g,
173        Err(poisoned) => poisoned.into_inner(),
174    };
175    registry.remove(id)
176}
177
178pub fn take_stream_observer(exchange: &Exchange) -> Option<GrpcStreamObserver> {
179    let id = exchange
180        .properties
181        .get("CamelGrpcStreamObserverId")?
182        .as_str()?;
183    remove_observer(id)
184}
185
186// ── Observer guard (auto-cleanup on Drop) ──────────────────────────────────
187
188struct ObserverGuard {
189    id: String,
190}
191
192impl ObserverGuard {
193    fn new(id: String) -> Self {
194        Self { id }
195    }
196}
197
198impl Drop for ObserverGuard {
199    fn drop(&mut self) {
200        remove_observer(&self.id);
201    }
202}
203
204// ── GrpcStreamObserver ─────────────────────────────────────────────────────
205
206#[derive(Clone)]
207pub struct GrpcStreamObserver {
208    tx: mpsc::Sender<GrpcStreamItem>,
209    resp_desc: MessageDescriptor,
210}
211
212impl GrpcStreamObserver {
213    pub(crate) fn new(tx: mpsc::Sender<GrpcStreamItem>, resp_desc: MessageDescriptor) -> Self {
214        Self { tx, resp_desc }
215    }
216
217    pub async fn on_next(&self, json: serde_json::Value) -> Result<(), CamelError> {
218        let encoded = json_to_protobuf_bytes(json, self.resp_desc.clone())
219            .map_err(|e| CamelError::ProcessorError(format!("failed to encode protobuf: {e}")))?;
220        self.tx
221            .send(GrpcStreamItem::Message(encoded))
222            .await
223            .map_err(|_| CamelError::ProcessorError("stream observer channel closed".into()))
224    }
225
226    pub async fn on_error(&self, status: Status) {
227        if self.tx.send(GrpcStreamItem::Error(status)).await.is_err() {
228            tracing::debug!("grpc stream observer: failed to send error, channel closed");
229        }
230    }
231
232    pub async fn on_completed(&self) {
233        if self.tx.send(GrpcStreamItem::Done).await.is_err() {
234            tracing::debug!("grpc stream observer: failed to send done, channel closed");
235        }
236    }
237}
238
239// ── Helper ─────────────────────────────────────────────────────────────────
240
241fn json_to_protobuf_bytes(
242    json: serde_json::Value,
243    desc: MessageDescriptor,
244) -> Result<Vec<u8>, Status> {
245    let json_str = serde_json::to_string(&json)
246        .map_err(|e| Status::internal(format!("failed to serialize JSON: {e}")))?;
247    let mut de = serde_json::Deserializer::from_str(&json_str);
248    let resp_dyn = DynamicMessage::deserialize(desc, &mut de)
249        .map_err(|e| Status::internal(format!("failed to parse JSON into protobuf: {e}")))?;
250    let mut buf = BytesMut::new();
251    prost::Message::encode(&resp_dyn, &mut buf)
252        .map_err(|e| Status::internal(format!("failed to encode protobuf: {e}")))?;
253    Ok(buf.to_vec())
254}
255
256// ── GrpcConsumer ───────────────────────────────────────────────────────────
257
258pub struct GrpcConsumer {
259    host: String,
260    port: u16,
261    path: String,
262    proto_path: PathBuf,
263    service_name: String,
264    method_name: String,
265    mode: GrpcMode,
266    security_ctx: Option<SecurityContext>,
267    runtime: Arc<dyn camel_component_api::RuntimeObservability>,
268    server_config: GrpcServerConfig,
269}
270
271impl GrpcConsumer {
272    #[allow(clippy::too_many_arguments)]
273    pub fn new(
274        host: String,
275        port: u16,
276        path: String,
277        proto_path: PathBuf,
278        service_name: String,
279        method_name: String,
280        mode: GrpcMode,
281        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
282        server_config: GrpcServerConfig,
283    ) -> Self {
284        Self {
285            host,
286            port,
287            path,
288            proto_path,
289            service_name,
290            method_name,
291            mode,
292            security_ctx: None,
293            runtime,
294            server_config,
295        }
296    }
297
298    fn resolve_descriptors(&self) -> Result<(MessageDescriptor, MessageDescriptor), CamelError> {
299        let cache = proto_cache();
300        let pool = cache
301            .get_or_compile(&self.proto_path, std::iter::empty::<&std::path::Path>())
302            .map_err(|e| {
303                CamelError::EndpointCreationFailed(format!("failed to compile proto: {e}"))
304            })?;
305
306        let svc = pool
307            .get_service_by_name(&self.service_name)
308            .ok_or_else(|| {
309                CamelError::EndpointCreationFailed(format!(
310                    "service descriptor not found: {}",
311                    self.service_name
312                ))
313            })?;
314
315        let method = svc
316            .methods()
317            .find(|m| m.name() == self.method_name)
318            .ok_or_else(|| {
319                CamelError::EndpointCreationFailed(format!(
320                    "method descriptor not found: {}/{}",
321                    self.service_name, self.method_name
322                ))
323            })?;
324
325        Ok((method.input(), method.output()))
326    }
327
328    pub async fn start_with_listener(
329        &mut self,
330        ctx: ConsumerContext,
331        listener: tokio::net::TcpListener,
332    ) -> Result<(), CamelError> {
333        let dispatch = GrpcServerRegistry::global()
334            .get_or_spawn_with_listener(
335                listener,
336                &self.host,
337                self.port,
338                self.server_config.clone(),
339                Arc::clone(&self.runtime),
340            )
341            .await?;
342        self.start_inner(ctx, dispatch).await
343    }
344
345    async fn start_inner(
346        &mut self,
347        ctx: ConsumerContext,
348        dispatch: GrpcDispatchTable,
349    ) -> Result<(), CamelError> {
350        let (req_desc, resp_desc) = self.resolve_descriptors()?;
351        let mode = self.mode;
352
353        let (env_tx, mut env_rx) = mpsc::channel::<GrpcRequestEnvelope>(64);
354        {
355            let mut table = dispatch.write().await;
356            if table.contains_key(&self.path) {
357                return Err(CamelError::EndpointCreationFailed(format!(
358                    "duplicate gRPC consumer path: {}",
359                    self.path
360                )));
361            }
362            let authenticator = self
363                .security_ctx
364                .as_ref()
365                .map(|ctx| ctx.authenticator.clone());
366            table.insert(self.path.clone(), (env_tx, mode, authenticator));
367        }
368
369        let path = self.path.clone();
370        let host = self.host.clone();
371        let port = self.port;
372        let sender = ctx.sender();
373
374        info!(
375            path = %path,
376            host = %host,
377            port = port,
378            mode = ?mode,
379            "grpc consumer started, waiting for requests"
380        );
381
382        // NOTE: Long-running bidi streams hold a semaphore permit for their duration.
383        // If this becomes an issue, consider separate concurrency limits for streaming vs unary.
384        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(64));
385        let mut join_set = tokio::task::JoinSet::new();
386
387        loop {
388            tokio::select! {
389                biased;
390                _ = ctx.cancelled() => {
391                    info!(
392                        path = %path,
393                        "grpc consumer cancelled, shutting down"
394                    );
395                    break;
396                }
397                envelope = env_rx.recv() => {
398                    let Some(envelope) = envelope else { break };
399
400                    let sem = semaphore.clone();
401                    let permit = sem.acquire_owned().await.map_err(|_| CamelError::ChannelClosed)?;
402                    let req_desc = req_desc.clone();
403                    let resp_desc = resp_desc.clone();
404                    let sender = sender.clone();
405                    let correlation_id = next_observer_id();
406                    let path_for_log = path.clone();
407                    let policy = self.security_ctx.as_ref().map(|ctx| ctx.policy.clone());
408
409                    debug!(
410                        path = %path_for_log,
411                        correlation_id = %correlation_id,
412                        "grpc consumer received request"
413                    );
414
415                    join_set.spawn(async move {
416                        let _permit = permit;
417                        match envelope {
418                            GrpcRequestEnvelope::Unary { metadata, body, reply_tx, principal } => {
419                                debug!(
420                                    path = %path_for_log,
421                                    correlation_id = %correlation_id,
422                                    size = body.len(),
423                                    "grpc consumer processing unary request"
424                                );
425
426                                if let (Some(principal), Some(policy)) = (&principal, &policy) {
427                                    let mut exchange = Exchange::new(Message::new(Body::Empty));
428                                    store_principal_properties(&mut exchange, principal);
429                                    match policy.evaluate(&mut exchange).await {
430                                        Ok(AuthorizationDecision::Granted { .. }) => {
431                                            tracing::debug!(path = %path_for_log, subject = %principal.subject, "gRPC request authorized");
432                                        }
433                                        Ok(AuthorizationDecision::Denied { reason, .. }) => {
434                                            tracing::warn!(path = %path_for_log, reason = %reason, "gRPC request denied");
435                                            let _ = reply_tx.send(GrpcReply::Err(tonic::Status::permission_denied(reason)));
436                                            return;
437                                        }
438                                        Err(e) => {
439                                            // log-policy: system-broken
440                                            tracing::error!(path = %path_for_log, error = %e, "gRPC policy evaluation error");
441                                            let _ = reply_tx.send(GrpcReply::Err(tonic::Status::internal(format!("authorization error: {e}"))));
442                                            return;
443                                        }
444                                    }
445                                }
446
447                                let result = process_unary_request(
448                                    body, metadata, req_desc, resp_desc, sender,
449                                ).await;
450                                let reply = match result {
451                                    Ok(bytes) => GrpcReply::Ok(bytes),
452                                    Err(status) => GrpcReply::Err(status),
453                                };
454                                let _ = reply_tx.send(reply);
455                            }
456                            GrpcRequestEnvelope::ServerStreaming { metadata, body, reply_tx, principal } => {
457                                debug!(
458                                    path = %path_for_log,
459                                    correlation_id = %correlation_id,
460                                    size = body.len(),
461                                    "grpc consumer processing server streaming request"
462                                );
463
464                                if let (Some(principal), Some(policy)) = (&principal, &policy) {
465                                    let mut exchange = Exchange::new(Message::new(Body::Empty));
466                                    store_principal_properties(&mut exchange, principal);
467                                    match policy.evaluate(&mut exchange).await {
468                                        Ok(AuthorizationDecision::Granted { .. }) => {
469                                            tracing::debug!(path = %path_for_log, subject = %principal.subject, "gRPC request authorized");
470                                        }
471                                        Ok(AuthorizationDecision::Denied { reason, .. }) => {
472                                            tracing::warn!(path = %path_for_log, reason = %reason, "gRPC request denied");
473                                            let _ = reply_tx.send(GrpcStreamItem::Error(tonic::Status::permission_denied(reason))).await;
474                                            return;
475                                        }
476                                        Err(e) => {
477                                            // log-policy: system-broken
478                                            tracing::error!(path = %path_for_log, error = %e, "gRPC policy evaluation error");
479                                            let _ = reply_tx.send(GrpcStreamItem::Error(tonic::Status::internal(format!("authorization error: {e}")))).await;
480                                            return;
481                                        }
482                                    }
483                                }
484
485                                process_server_streaming_request(
486                                    body, metadata, req_desc, resp_desc, sender, reply_tx,
487                                ).await;
488                            }
489                            GrpcRequestEnvelope::ClientStreaming { metadata, body_rx, reply_tx, principal } => {
490                                debug!(
491                                    path = %path_for_log,
492                                    correlation_id = %correlation_id,
493                                    "grpc consumer processing client streaming request"
494                                );
495
496                                if let (Some(principal), Some(policy)) = (&principal, &policy) {
497                                    let mut exchange = Exchange::new(Message::new(Body::Empty));
498                                    store_principal_properties(&mut exchange, principal);
499                                    match policy.evaluate(&mut exchange).await {
500                                        Ok(AuthorizationDecision::Granted { .. }) => {
501                                            tracing::debug!(path = %path_for_log, subject = %principal.subject, "gRPC request authorized");
502                                        }
503                                        Ok(AuthorizationDecision::Denied { reason, .. }) => {
504                                            tracing::warn!(path = %path_for_log, reason = %reason, "gRPC request denied");
505                                            let _ = reply_tx.send(GrpcReply::Err(tonic::Status::permission_denied(reason)));
506                                            return;
507                                        }
508                                        Err(e) => {
509                                            // log-policy: system-broken
510                                            tracing::error!(path = %path_for_log, error = %e, "gRPC policy evaluation error");
511                                            let _ = reply_tx.send(GrpcReply::Err(tonic::Status::internal(format!("authorization error: {e}"))));
512                                            return;
513                                        }
514                                    }
515                                }
516
517                                process_client_streaming_request(
518                                    body_rx, metadata, req_desc, resp_desc, sender, reply_tx,
519                                ).await;
520                            }
521                            GrpcRequestEnvelope::Bidi { metadata, body_rx, reply_tx, principal } => {
522                                debug!(
523                                    path = %path_for_log,
524                                    correlation_id = %correlation_id,
525                                    "grpc consumer processing bidi streaming request"
526                                );
527
528                                if let (Some(principal), Some(policy)) = (&principal, &policy) {
529                                    let mut exchange = Exchange::new(Message::new(Body::Empty));
530                                    store_principal_properties(&mut exchange, principal);
531                                    match policy.evaluate(&mut exchange).await {
532                                        Ok(AuthorizationDecision::Granted { .. }) => {
533                                            tracing::debug!(path = %path_for_log, subject = %principal.subject, "gRPC request authorized");
534                                        }
535                                        Ok(AuthorizationDecision::Denied { reason, .. }) => {
536                                            tracing::warn!(path = %path_for_log, reason = %reason, "gRPC request denied");
537                                            let _ = reply_tx.send(GrpcStreamItem::Error(tonic::Status::permission_denied(reason))).await;
538                                            return;
539                                        }
540                                        Err(e) => {
541                                            // log-policy: system-broken
542                                            tracing::error!(path = %path_for_log, error = %e, "gRPC policy evaluation error");
543                                            let _ = reply_tx.send(GrpcStreamItem::Error(tonic::Status::internal(format!("authorization error: {e}")))).await;
544                                            return;
545                                        }
546                                    }
547                                }
548
549                                process_bidi_request(
550                                    body_rx, metadata, req_desc, resp_desc, sender, reply_tx,
551                                ).await;
552                            }
553                        }
554                    });
555                }
556            }
557        }
558
559        join_set.shutdown().await;
560
561        GrpcServerRegistry::global()
562            .unregister(&host, port, &path)
563            .await;
564
565        info!(
566            path = %path,
567            "grpc consumer stopped"
568        );
569
570        Ok(())
571    }
572}
573
574#[async_trait]
575impl Consumer for GrpcConsumer {
576    async fn start(&mut self, ctx: ConsumerContext) -> Result<(), CamelError> {
577        info!(
578            host = %self.host,
579            port = self.port,
580            service = %self.service_name,
581            method = %self.method_name,
582            mode = ?self.mode,
583            "grpc consumer starting"
584        );
585        let dispatch = GrpcServerRegistry::global()
586            .get_or_spawn(
587                &self.host,
588                self.port,
589                self.server_config.clone(),
590                Arc::clone(&self.runtime),
591            )
592            .await?;
593        // gRPC listener is bound inside get_or_spawn (TcpListener::bind
594        // before tokio::spawn). Signal readiness now that the bind succeeded.
595        ctx.mark_ready();
596        self.start_inner(ctx, dispatch).await
597    }
598
599    async fn stop(&mut self) -> Result<(), CamelError> {
600        info!(
601            host = %self.host,
602            port = self.port,
603            service = %self.service_name,
604            method = %self.method_name,
605            "grpc consumer stopping"
606        );
607        GrpcServerRegistry::global()
608            .unregister(&self.host, self.port, &self.path)
609            .await;
610        Ok(())
611    }
612
613    fn concurrency_model(&self) -> ConcurrencyModel {
614        ConcurrencyModel::Concurrent { max: None }
615    }
616
617    fn startup_mode(&self) -> ConsumerStartupMode {
618        ConsumerStartupMode::Explicit
619    }
620
621    fn set_security_context(&mut self, ctx: SecurityContext) {
622        self.security_ctx = Some(ctx);
623    }
624}
625
626// ── Unary processor (unchanged) ────────────────────────────────────────────
627
628async fn process_unary_request(
629    body: Vec<u8>,
630    metadata: tonic::metadata::MetadataMap,
631    req_desc: MessageDescriptor,
632    resp_desc: MessageDescriptor,
633    sender: mpsc::Sender<ExchangeEnvelope>,
634) -> Result<Vec<u8>, Status> {
635    let req_dyn = DynamicMessage::decode(req_desc, body.as_slice())
636        .map_err(|e| Status::invalid_argument(format!("failed to decode protobuf: {e}")))?;
637
638    let json = serde_json::to_value(&req_dyn).map_err(|e| {
639        Status::invalid_argument(format!("failed to convert protobuf to JSON: {e}"))
640    })?;
641
642    let mut msg = Message::new(Body::Json(json));
643    for (k, v) in extract_metadata(&metadata) {
644        msg.set_header(k, v);
645    }
646
647    let exchange = Exchange::new(msg);
648
649    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
650    let envelope = ExchangeEnvelope {
651        exchange,
652        reply_tx: Some(reply_tx),
653    };
654
655    sender
656        .send(envelope)
657        .await
658        .map_err(|_| Status::internal("pipeline channel closed"))?;
659
660    let result = reply_rx
661        .await
662        .map_err(|_| Status::internal("pipeline reply dropped"))?
663        .map_err(|e| Status::internal(format!("pipeline error: {e}")))?;
664
665    let resp_json = match result.input.body {
666        Body::Json(v) => v,
667        other => {
668            return Err(Status::internal(format!(
669                "expected JSON response body from pipeline, got {other:?}"
670            )));
671        }
672    };
673
674    let json_str = serde_json::to_string(&resp_json)
675        .map_err(|e| Status::internal(format!("failed to serialize response JSON: {e}")))?;
676    let mut de = serde_json::Deserializer::from_str(&json_str);
677    let resp_dyn = DynamicMessage::deserialize(resp_desc, &mut de)
678        .map_err(|e| Status::internal(format!("failed to parse JSON into protobuf: {e}")))?;
679
680    let mut buf = BytesMut::new();
681    resp_dyn
682        .encode(&mut buf)
683        .map_err(|e| Status::internal(format!("failed to encode protobuf response: {e}")))?;
684
685    Ok(buf.to_vec())
686}
687
688// ── Server-streaming processor ─────────────────────────────────────────────
689
690async fn process_server_streaming_request(
691    body: Vec<u8>,
692    metadata: tonic::metadata::MetadataMap,
693    req_desc: MessageDescriptor,
694    resp_desc: MessageDescriptor,
695    sender: mpsc::Sender<ExchangeEnvelope>,
696    reply_tx: mpsc::Sender<GrpcStreamItem>,
697) {
698    let req_dyn = match DynamicMessage::decode(req_desc, body.as_slice()) {
699        Ok(m) => m,
700        Err(e) => {
701            let _ = reply_tx
702                .send(GrpcStreamItem::Error(Status::invalid_argument(format!(
703                    "failed to decode protobuf: {e}"
704                ))))
705                .await;
706            return;
707        }
708    };
709
710    let json = match serde_json::to_value(&req_dyn) {
711        Ok(v) => v,
712        Err(e) => {
713            let _ = reply_tx
714                .send(GrpcStreamItem::Error(Status::invalid_argument(format!(
715                    "failed to convert protobuf to JSON: {e}"
716                ))))
717                .await;
718            return;
719        }
720    };
721
722    let mut msg = Message::new(Body::Json(json));
723    for (k, v) in extract_metadata(&metadata) {
724        msg.set_header(k, v);
725    }
726
727    let observer = GrpcStreamObserver::new(reply_tx.clone(), resp_desc);
728    let observer_id = next_observer_id();
729    register_observer(observer_id.clone(), observer);
730    let _guard = ObserverGuard::new(observer_id.clone());
731
732    let mut exchange = Exchange::new(msg);
733    exchange.set_property("CamelGrpcStreamObserverId", Value::String(observer_id));
734
735    let envelope = ExchangeEnvelope {
736        exchange,
737        reply_tx: None,
738    };
739
740    if sender.send(envelope).await.is_err() {
741        let _ = reply_tx
742            .send(GrpcStreamItem::Error(Status::internal(
743                "pipeline channel closed",
744            )))
745            .await;
746    }
747
748    // Wait for the stream receiver to be dropped (stream complete).
749    // This keeps the guard alive so the observer stays registered until
750    // the route is done. If take_stream_observer was called, the guard's
751    // Drop is a no-op. If not, the guard cleans up the leaked observer.
752    reply_tx.closed().await;
753}
754
755// ── Client-streaming processor ─────────────────────────────────────────────
756
757async fn process_client_streaming_request(
758    mut body_rx: mpsc::Receiver<Vec<u8>>,
759    metadata: tonic::metadata::MetadataMap,
760    req_desc: MessageDescriptor,
761    resp_desc: MessageDescriptor,
762    sender: mpsc::Sender<ExchangeEnvelope>,
763    reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
764) {
765    while let Some(body) = body_rx.recv().await {
766        let req_dyn = match DynamicMessage::decode(req_desc.clone(), body.as_slice()) {
767            Ok(d) => d,
768            Err(e) => {
769                let _ = reply_tx.send(GrpcReply::Err(Status::invalid_argument(format!(
770                    "failed to decode protobuf: {e}"
771                ))));
772                return;
773            }
774        };
775
776        let json = match serde_json::to_value(&req_dyn) {
777            Ok(j) => j,
778            Err(e) => {
779                let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
780                    "failed to convert protobuf to JSON: {e}"
781                ))));
782                return;
783            }
784        };
785
786        let mut msg = Message::new(Body::Json(json));
787        for (k, v) in extract_metadata(&metadata) {
788            msg.set_header(k, v);
789        }
790        msg.set_header(
791            "CamelGrpcClientStreaming".to_string(),
792            serde_json::Value::Bool(true),
793        );
794
795        let exchange = Exchange::new(msg);
796        let (reply_tx_pipe, reply_rx_pipe) = tokio::sync::oneshot::channel();
797        let envelope = ExchangeEnvelope {
798            exchange,
799            reply_tx: Some(reply_tx_pipe),
800        };
801
802        if sender.send(envelope).await.is_err() {
803            let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline channel closed")));
804            return;
805        }
806
807        // Intentionally discard intermediate replies — only the completion exchange's reply matters.
808        let _ = reply_rx_pipe.await;
809    }
810
811    // Stream complete — send final Exchange with completion marker
812    let mut completion_msg = Message::new(Body::Json(serde_json::Value::Null));
813    for (k, v) in extract_metadata(&metadata) {
814        completion_msg.set_header(k, v);
815    }
816    completion_msg.set_header(
817        "CamelGrpcClientStreaming".to_string(),
818        serde_json::Value::Bool(true),
819    );
820    completion_msg.set_header(
821        "CamelGrpcClientStreamComplete".to_string(),
822        serde_json::Value::Bool(true),
823    );
824
825    let completion_exchange = Exchange::new(completion_msg);
826    let (reply_tx_pipe, reply_rx_pipe) = tokio::sync::oneshot::channel();
827    let envelope = ExchangeEnvelope {
828        exchange: completion_exchange,
829        reply_tx: Some(reply_tx_pipe),
830    };
831
832    if sender.send(envelope).await.is_err() {
833        let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline channel closed")));
834        return;
835    }
836
837    // The route's response to the completion Exchange becomes the gRPC response
838    let result = match reply_rx_pipe.await {
839        Ok(Ok(exchange)) => exchange,
840        Ok(Err(e)) => {
841            let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
842                "pipeline error: {e}"
843            ))));
844            return;
845        }
846        Err(_) => {
847            let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline reply dropped")));
848            return;
849        }
850    };
851
852    let resp_json = match result.input.body {
853        Body::Json(v) => v,
854        other => {
855            let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
856                "expected JSON response body from pipeline, got {other:?}"
857            ))));
858            return;
859        }
860    };
861
862    let encoded = match json_to_protobuf_bytes(resp_json, resp_desc) {
863        Ok(b) => b,
864        Err(e) => {
865            let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
866                "failed to encode response: {e}",
867            ))));
868            return;
869        }
870    };
871
872    let _ = reply_tx.send(GrpcReply::Ok(encoded));
873}
874
875// ── Bidi-streaming processor ───────────────────────────────────────────────
876
877async fn process_bidi_request(
878    mut body_rx: mpsc::Receiver<Vec<u8>>,
879    metadata: tonic::metadata::MetadataMap,
880    req_desc: MessageDescriptor,
881    resp_desc: MessageDescriptor,
882    sender: mpsc::Sender<ExchangeEnvelope>,
883    reply_tx: mpsc::Sender<GrpcStreamItem>,
884) {
885    let observer = GrpcStreamObserver::new(reply_tx.clone(), resp_desc);
886    let observer_id = next_observer_id();
887    register_observer(observer_id.clone(), observer.clone());
888    let _guard = ObserverGuard::new(observer_id.clone());
889
890    // Spawn a task to forward messages from the client stream to the pipeline
891    let sender_clone = sender.clone();
892    let metadata_clone = metadata.clone();
893    let req_desc_clone = req_desc.clone();
894
895    let forward_task = tokio::spawn(async move {
896        let mut sequence: u64 = 0;
897        while let Some(body) = body_rx.recv().await {
898            let req_dyn = match DynamicMessage::decode(req_desc_clone.clone(), body.as_slice()) {
899                Ok(m) => m,
900                Err(e) => {
901                    let _ = observer
902                        .on_error(Status::invalid_argument(format!(
903                            "failed to decode protobuf: {e}"
904                        )))
905                        .await;
906                    continue;
907                }
908            };
909
910            let json = match serde_json::to_value(&req_dyn) {
911                Ok(v) => v,
912                Err(e) => {
913                    let _ = observer
914                        .on_error(Status::invalid_argument(format!(
915                            "failed to convert protobuf to JSON: {e}"
916                        )))
917                        .await;
918                    continue;
919                }
920            };
921
922            let mut msg = Message::new(Body::Json(json));
923            for (k, v) in extract_metadata(&metadata_clone) {
924                msg.set_header(k, v);
925            }
926
927            msg.set_header(
928                "CamelGrpcBidiSequence",
929                serde_json::Value::Number(sequence.into()),
930            );
931            sequence += 1;
932
933            let mut exchange = Exchange::new(msg);
934            exchange.set_property(
935                "CamelGrpcStreamObserverId",
936                Value::String(observer_id.clone()),
937            );
938
939            let envelope = ExchangeEnvelope {
940                exchange,
941                reply_tx: None,
942            };
943
944            if sender_clone.send(envelope).await.is_err() {
945                let _ = observer
946                    .on_error(Status::internal("pipeline channel closed"))
947                    .await;
948                break;
949            }
950        }
951
952        // Signal completion when client stream ends
953        observer.on_completed().await;
954    });
955
956    // Wait for the forward task to complete
957    let _ = forward_task.await;
958}
camel_component_grpc/consumer.rs

camel_component_grpc/
consumer.rs
