oxirs_stream/
bridge.rs

1//! # Message Queue Bridge Module
2//!
3//! This module provides comprehensive message queue integration for external systems:
4//! - Protocol bridging between different message queue systems
5//! - Format conversion and message transformation
6//! - Routing rules and message filtering
7//! - External system adapters
8//! - Performance monitoring and diagnostics
9
10use anyhow::{anyhow, Result};
11use serde::{Deserialize, Serialize};
12use std::collections::{HashMap, VecDeque};
13use std::sync::Arc;
14use std::time::{Duration, Instant};
15use tokio::sync::{broadcast, RwLock};
16use tokio::time::interval;
17use tracing::{debug, error, info, warn};
18use uuid::Uuid;
19
20/// Message queue bridge manager
21pub struct MessageBridgeManager {
22    /// Registered bridges
23    bridges: Arc<RwLock<HashMap<String, MessageBridge>>>,
24    /// Bridge configurations
25    configs: Arc<RwLock<HashMap<String, BridgeConfig>>>,
26    /// Message transformers
27    transformers: Arc<RwLock<HashMap<String, Box<dyn MessageTransformer + Send + Sync>>>>,
28    /// Routing engine
29    router: Arc<RoutingEngine>,
30    /// Statistics
31    stats: Arc<RwLock<BridgeStats>>,
32    /// Event notifier
33    event_notifier: broadcast::Sender<BridgeNotification>,
34}
35
36/// Message bridge
37#[derive(Clone)]
38struct MessageBridge {
39    /// Bridge ID
40    id: String,
41    /// Bridge type
42    bridge_type: BridgeType,
43    /// Source configuration
44    source: ExternalSystemConfig,
45    /// Target configuration
46    target: ExternalSystemConfig,
47    /// Message transformer
48    transformer: String,
49    /// Routing rules
50    routing_rules: Vec<RoutingRule>,
51    /// Bridge status
52    status: BridgeStatus,
53    /// Statistics
54    stats: BridgeStatistics,
55    /// Created timestamp
56    created_at: Instant,
57    /// Last activity
58    last_activity: Option<Instant>,
59}
60
61/// Bridge types
62#[derive(Debug, Clone, Serialize, Deserialize)]
63pub enum BridgeType {
64    /// Bidirectional bridge
65    Bidirectional,
66    /// Source to target only
67    SourceToTarget,
68    /// Target to source only
69    TargetToSource,
70    /// Fanout (one source, multiple targets)
71    Fanout,
72    /// Fanin (multiple sources, one target)
73    Fanin,
74}
75
76/// External system configuration
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct ExternalSystemConfig {
79    /// System type
80    pub system_type: ExternalSystemType,
81    /// Connection configuration
82    pub connection: ConnectionConfig,
83    /// Format configuration
84    pub format: FormatConfig,
85    /// Security configuration
86    pub security: SecurityConfig,
87}
88
89/// External system types
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub enum ExternalSystemType {
92    /// Apache Kafka
93    Kafka {
94        brokers: Vec<String>,
95        topics: Vec<String>,
96        consumer_group: Option<String>,
97    },
98    /// RabbitMQ
99    RabbitMQ {
100        url: String,
101        exchange: String,
102        routing_key: String,
103        queue: Option<String>,
104    },
105    /// Amazon SQS
106    AmazonSQS {
107        region: String,
108        queue_url: String,
109        credentials: AwsCredentials,
110    },
111    /// Azure Service Bus
112    AzureServiceBus {
113        connection_string: String,
114        queue_name: String,
115    },
116    /// Google Cloud Pub/Sub
117    GooglePubSub {
118        project_id: String,
119        topic: String,
120        subscription: Option<String>,
121    },
122    /// Apache Pulsar
123    Pulsar {
124        service_url: String,
125        topics: Vec<String>,
126        subscription: Option<String>,
127    },
128    /// Redis Pub/Sub
129    RedisPubSub { url: String, channels: Vec<String> },
130    /// HTTP REST API
131    HttpRest {
132        base_url: String,
133        endpoints: HashMap<String, String>,
134        headers: HashMap<String, String>,
135    },
136    /// WebSocket
137    WebSocket { url: String, protocols: Vec<String> },
138    /// File system
139    FileSystem {
140        directory: String,
141        pattern: String,
142        watch_mode: bool,
143    },
144    /// MQTT broker
145    Mqtt {
146        broker_url: String,
147        client_id: String,
148        topic_subscriptions: Vec<String>,
149        qos: u8,
150        username: Option<String>,
151        password: Option<String>,
152    },
153    /// OPC UA server
154    OpcUa {
155        endpoint_url: String,
156        security_policy: String,
157        user_identity: String,
158        node_subscriptions: Vec<String>,
159    },
160    /// Eclipse Sparkplug B (MQTT-based Industry 4.0)
161    SparkplugB {
162        broker_url: String,
163        group_id: String,
164        edge_node_id: String,
165        device_ids: Vec<String>,
166    },
167}
168
169/// AWS credentials
170#[derive(Debug, Clone, Serialize, Deserialize)]
171pub struct AwsCredentials {
172    pub access_key_id: String,
173    pub secret_access_key: String,
174    pub session_token: Option<String>,
175}
176
177/// Connection configuration
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct ConnectionConfig {
180    /// Connection timeout
181    pub timeout: Duration,
182    /// Keep alive interval
183    pub keep_alive: Duration,
184    /// Retry configuration
185    pub retry: RetryConfig,
186    /// SSL/TLS configuration
187    pub tls: Option<TlsConfig>,
188}
189
190/// TLS configuration
191#[derive(Debug, Clone, Serialize, Deserialize)]
192pub struct TlsConfig {
193    pub enabled: bool,
194    pub verify_certificate: bool,
195    pub certificate_path: Option<String>,
196    pub private_key_path: Option<String>,
197    pub ca_certificate_path: Option<String>,
198}
199
200/// Retry configuration
201#[derive(Debug, Clone, Serialize, Deserialize)]
202pub struct RetryConfig {
203    pub max_attempts: u32,
204    pub initial_delay: Duration,
205    pub max_delay: Duration,
206    pub exponential_backoff: bool,
207}
208
209/// Format configuration
210#[derive(Debug, Clone, Serialize, Deserialize)]
211pub struct FormatConfig {
212    /// Message format
213    pub format: MessageFormat,
214    /// Encoding
215    pub encoding: String,
216    /// Compression
217    pub compression: Option<CompressionType>,
218    /// Schema validation
219    pub schema_validation: bool,
220}
221
222/// Message formats
223#[derive(Debug, Clone, Serialize, Deserialize)]
224pub enum MessageFormat {
225    /// JSON format
226    Json,
227    /// Apache Avro
228    Avro { schema: String },
229    /// Protocol Buffers
230    Protobuf { schema: String },
231    /// XML format
232    Xml,
233    /// Plain text
234    Text,
235    /// Binary format
236    Binary,
237    /// Custom format
238    Custom { transformer: String },
239}
240
241/// Compression types
242#[derive(Debug, Clone, Serialize, Deserialize)]
243pub enum CompressionType {
244    Gzip,
245    Snappy,
246    Lz4,
247    Zstd,
248}
249
250/// Security configuration
251#[derive(Debug, Clone, Serialize, Deserialize)]
252pub struct SecurityConfig {
253    /// Authentication method
254    pub auth: AuthenticationMethod,
255    /// Encryption settings
256    pub encryption: EncryptionConfig,
257    /// Access control
258    pub access_control: AccessControlConfig,
259}
260
261/// Authentication methods
262#[derive(Debug, Clone, Serialize, Deserialize)]
263pub enum AuthenticationMethod {
264    None,
265    BasicAuth {
266        username: String,
267        password: String,
268    },
269    BearerToken {
270        token: String,
271    },
272    ApiKey {
273        key: String,
274        header: String,
275    },
276    OAuth2 {
277        client_id: String,
278        client_secret: String,
279        token_url: String,
280    },
281    SaslPlain {
282        username: String,
283        password: String,
284    },
285    SaslScramSha256 {
286        username: String,
287        password: String,
288    },
289    Certificate {
290        cert_path: String,
291        key_path: String,
292    },
293}
294
295/// Encryption configuration
296#[derive(Debug, Clone, Serialize, Deserialize)]
297pub struct EncryptionConfig {
298    pub enabled: bool,
299    pub algorithm: Option<String>,
300    pub key_id: Option<String>,
301}
302
303/// Access control configuration
304#[derive(Debug, Clone, Serialize, Deserialize)]
305pub struct AccessControlConfig {
306    pub read_permissions: Vec<String>,
307    pub write_permissions: Vec<String>,
308    pub admin_permissions: Vec<String>,
309}
310
311/// Routing rule
312#[derive(Debug, Clone, Serialize, Deserialize)]
313pub struct RoutingRule {
314    /// Rule name
315    pub name: String,
316    /// Rule condition
317    pub condition: RuleCondition,
318    /// Rule action
319    pub action: RuleAction,
320    /// Rule priority
321    pub priority: u32,
322    /// Rule enabled
323    pub enabled: bool,
324}
325
326/// Rule condition
327#[derive(Debug, Clone, Serialize, Deserialize)]
328pub enum RuleCondition {
329    /// Always match
330    Always,
331    /// Match event type
332    EventType { types: Vec<String> },
333    /// Match graph
334    Graph { patterns: Vec<String> },
335    /// Match subject pattern
336    SubjectPattern { regex: String },
337    /// Match predicate
338    Predicate { predicates: Vec<String> },
339    /// Custom expression
340    Expression { expr: String },
341    /// Composite condition
342    Composite {
343        operator: LogicalOperator,
344        conditions: Vec<RuleCondition>,
345    },
346}
347
348/// Logical operators
349#[derive(Debug, Clone, Serialize, Deserialize)]
350pub enum LogicalOperator {
351    And,
352    Or,
353    Not,
354}
355
356/// Rule action
357#[derive(Debug, Clone, Serialize, Deserialize)]
358pub enum RuleAction {
359    /// Forward message
360    Forward,
361    /// Drop message
362    Drop,
363    /// Transform message
364    Transform { transformer: String },
365    /// Route to specific target
366    Route { target: String },
367    /// Duplicate message
368    Duplicate { targets: Vec<String> },
369}
370
371/// Bridge status
372#[derive(Debug, Clone, PartialEq)]
373enum BridgeStatus {
374    Active,
375    #[allow(dead_code)]
376    Paused,
377    Stopped,
378    #[allow(dead_code)]
379    Failed {
380        reason: String,
381    },
382}
383
384/// Bridge configuration
385#[derive(Debug, Clone)]
386pub struct BridgeConfig {
387    /// Maximum message queue size
388    pub max_queue_size: usize,
389    /// Batch size for processing
390    pub batch_size: usize,
391    /// Processing interval
392    pub processing_interval: Duration,
393    /// Enable monitoring
394    pub enable_monitoring: bool,
395    /// Enable dead letter queue
396    pub enable_dlq: bool,
397    /// Message TTL
398    pub message_ttl: Duration,
399}
400
401impl Default for BridgeConfig {
402    fn default() -> Self {
403        Self {
404            max_queue_size: 10000,
405            batch_size: 100,
406            processing_interval: Duration::from_millis(100),
407            enable_monitoring: true,
408            enable_dlq: true,
409            message_ttl: Duration::from_secs(24 * 60 * 60),
410        }
411    }
412}
413
414/// Bridge statistics
415#[derive(Debug, Clone, Default)]
416pub struct BridgeStatistics {
417    /// Messages received
418    pub messages_received: u64,
419    /// Messages sent
420    pub messages_sent: u64,
421    /// Messages dropped
422    pub messages_dropped: u64,
423    /// Messages failed
424    pub messages_failed: u64,
425    /// Transform errors
426    pub transform_errors: u64,
427    /// Average processing time
428    pub avg_processing_time: Duration,
429    /// Last activity
430    pub last_activity: Option<Instant>,
431}
432
433/// Manager statistics
434#[derive(Debug, Clone, Default)]
435pub struct BridgeStats {
436    /// Total bridges
437    pub total_bridges: usize,
438    /// Active bridges
439    pub active_bridges: usize,
440    /// Total messages processed
441    pub total_messages: u64,
442    /// Failed messages
443    pub failed_messages: u64,
444    /// Average processing time
445    pub avg_processing_time: Duration,
446}
447
448/// Bridge notification events
449#[derive(Debug, Clone)]
450pub enum BridgeNotification {
451    /// Bridge created
452    BridgeCreated { id: String, bridge_type: BridgeType },
453    /// Bridge started
454    BridgeStarted { id: String },
455    /// Bridge stopped
456    BridgeStopped { id: String },
457    /// Bridge failed
458    BridgeFailed { id: String, reason: String },
459    /// Message processed
460    MessageProcessed {
461        bridge_id: String,
462        message_id: String,
463        duration: Duration,
464    },
465    /// Message failed
466    MessageFailed {
467        bridge_id: String,
468        message_id: String,
469        error: String,
470    },
471}
472
473/// Message transformer trait
474pub trait MessageTransformer {
475    /// Transform message from source format to target format
476    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage>;
477
478    /// Get transformer name
479    fn name(&self) -> &str;
480
481    /// Get supported formats
482    fn supported_formats(&self) -> (MessageFormat, MessageFormat);
483}
484
485/// External message representation
486#[derive(Debug, Clone, Serialize, Deserialize)]
487pub struct ExternalMessage {
488    /// Message ID
489    pub id: String,
490    /// Message headers
491    pub headers: HashMap<String, String>,
492    /// Message payload
493    pub payload: Vec<u8>,
494    /// Message format
495    pub format: MessageFormat,
496    /// Timestamp
497    pub timestamp: chrono::DateTime<chrono::Utc>,
498    /// Source system
499    pub source: String,
500    /// Message metadata
501    pub metadata: HashMap<String, String>,
502}
503
504/// Routing engine
505struct RoutingEngine {
506    /// Global routing rules
507    _global_rules: Arc<RwLock<Vec<RoutingRule>>>,
508    /// Bridge-specific rules cache
509    _rule_cache: Arc<RwLock<HashMap<String, Vec<RoutingRule>>>>,
510}
511
512impl MessageBridgeManager {
513    /// Create a new message bridge manager
514    pub async fn new() -> Result<Self> {
515        let (tx, _) = broadcast::channel(1000);
516
517        Ok(Self {
518            bridges: Arc::new(RwLock::new(HashMap::new())),
519            configs: Arc::new(RwLock::new(HashMap::new())),
520            transformers: Arc::new(RwLock::new(HashMap::new())),
521            router: Arc::new(RoutingEngine::new()),
522            stats: Arc::new(RwLock::new(BridgeStats::default())),
523            event_notifier: tx,
524        })
525    }
526
527    /// Register a message transformer
528    pub async fn register_transformer(
529        &self,
530        transformer: Box<dyn MessageTransformer + Send + Sync>,
531    ) {
532        let name = transformer.name().to_string();
533        self.transformers.write().await.insert(name, transformer);
534        info!("Registered message transformer");
535    }
536
537    /// Create a message bridge
538    pub async fn create_bridge(
539        &self,
540        bridge_type: BridgeType,
541        source: ExternalSystemConfig,
542        target: ExternalSystemConfig,
543        transformer: String,
544        routing_rules: Vec<RoutingRule>,
545        config: BridgeConfig,
546    ) -> Result<String> {
547        // Validate transformer exists
548        if !self.transformers.read().await.contains_key(&transformer) {
549            return Err(anyhow!("Transformer not found: {}", transformer));
550        }
551
552        // Generate bridge ID
553        let bridge_id = Uuid::new_v4().to_string();
554
555        // Create bridge
556        let bridge = MessageBridge {
557            id: bridge_id.clone(),
558            bridge_type: bridge_type.clone(),
559            source,
560            target,
561            transformer,
562            routing_rules,
563            status: BridgeStatus::Stopped,
564            stats: BridgeStatistics::default(),
565            created_at: Instant::now(),
566            last_activity: None,
567        };
568
569        // Register bridge
570        self.bridges.write().await.insert(bridge_id.clone(), bridge);
571        self.configs.write().await.insert(bridge_id.clone(), config);
572
573        // Update statistics
574        let mut stats = self.stats.write().await;
575        stats.total_bridges += 1;
576        drop(stats);
577
578        // Notify
579        let _ = self.event_notifier.send(BridgeNotification::BridgeCreated {
580            id: bridge_id.clone(),
581            bridge_type,
582        });
583
584        info!("Created message bridge: {}", bridge_id);
585        Ok(bridge_id)
586    }
587
588    /// Whether a real transport is implemented for the given external system.
589    ///
590    /// Only the FileSystem transport is implemented in pure Rust today. Network
591    /// transports (Kafka/RabbitMQ/Redis/HTTP/etc.) are not, so bridges using
592    /// them must fail loud at start rather than reporting Active while silently
593    /// transferring zero messages.
594    fn transport_implemented(system_type: &ExternalSystemType) -> bool {
595        matches!(system_type, ExternalSystemType::FileSystem { .. })
596    }
597
598    /// Start a bridge
599    pub async fn start_bridge(&self, bridge_id: &str) -> Result<()> {
600        // Validate that both endpoints have an implemented transport BEFORE
601        // marking the bridge active. This upholds the fail-loud contract: a
602        // bridge to an unimplemented backend must error, not silently no-op.
603        {
604            let bridges = self.bridges.read().await;
605            let bridge = bridges
606                .get(bridge_id)
607                .ok_or_else(|| anyhow!("Bridge not found"))?;
608
609            if !Self::transport_implemented(&bridge.source.system_type) {
610                return Err(anyhow!(
611                    "Bridge source transport {:?} is not implemented; cannot start bridge {}",
612                    bridge.source.system_type,
613                    bridge_id
614                ));
615            }
616            if !Self::transport_implemented(&bridge.target.system_type) {
617                return Err(anyhow!(
618                    "Bridge target transport {:?} is not implemented; cannot start bridge {}",
619                    bridge.target.system_type,
620                    bridge_id
621                ));
622            }
623        }
624
625        let bridge_exists = {
626            let mut bridges = self.bridges.write().await;
627            if let Some(bridge) = bridges.get_mut(bridge_id) {
628                bridge.status = BridgeStatus::Active;
629                true
630            } else {
631                false
632            }
633        };
634
635        if !bridge_exists {
636            return Err(anyhow!("Bridge not found"));
637        }
638
639        // Start bridge processing
640        self.start_bridge_processing(bridge_id).await?;
641
642        // Update statistics
643        self.stats.write().await.active_bridges += 1;
644
645        // Notify
646        let _ = self.event_notifier.send(BridgeNotification::BridgeStarted {
647            id: bridge_id.to_string(),
648        });
649
650        info!("Started bridge: {}", bridge_id);
651        Ok(())
652    }
653
654    /// Stop a bridge
655    pub async fn stop_bridge(&self, bridge_id: &str) -> Result<()> {
656        let mut bridges = self.bridges.write().await;
657        let bridge = bridges
658            .get_mut(bridge_id)
659            .ok_or_else(|| anyhow!("Bridge not found"))?;
660
661        bridge.status = BridgeStatus::Stopped;
662
663        // Update statistics
664        self.stats.write().await.active_bridges = bridges
665            .values()
666            .filter(|b| b.status == BridgeStatus::Active)
667            .count();
668
669        // Notify
670        let _ = self.event_notifier.send(BridgeNotification::BridgeStopped {
671            id: bridge_id.to_string(),
672        });
673
674        info!("Stopped bridge: {}", bridge_id);
675        Ok(())
676    }
677
678    /// Start bridge processing
679    async fn start_bridge_processing(&self, bridge_id: &str) -> Result<()> {
680        // Clone all necessary data before spawning the task
681        let bridge = {
682            let bridges_guard = self.bridges.read().await;
683            bridges_guard
684                .get(bridge_id)
685                .ok_or_else(|| anyhow!("Bridge not found"))?
686                .clone()
687        };
688
689        let config = {
690            let configs_guard = self.configs.read().await;
691            configs_guard
692                .get(bridge_id)
693                .ok_or_else(|| anyhow!("Bridge config not found"))?
694                .clone()
695        };
696
697        let bridges = self.bridges.clone();
698        let transformers = self.transformers.clone();
699        let router = self.router.clone();
700        let stats = self.stats.clone();
701        let event_notifier = self.event_notifier.clone();
702        let bridge_id = bridge_id.to_string();
703
704        tokio::spawn(async move {
705            let mut interval = interval(config.processing_interval);
706            let mut message_queue = VecDeque::new();
707
708            loop {
709                interval.tick().await;
710
711                // Check if bridge is still active
712                let status = {
713                    let bridges_guard = bridges.read().await;
714                    bridges_guard.get(&bridge_id).map(|b| b.status.clone())
715                };
716
717                if let Some(BridgeStatus::Active) = status {
718                    // Process messages from source
719                    match MessageBridgeManager::receive_messages(&bridge.source, &config).await {
720                        Ok(messages) => {
721                            for message in messages {
722                                message_queue.push_back(message);
723
724                                // Limit queue size
725                                if message_queue.len() > config.max_queue_size {
726                                    message_queue.pop_front();
727                                    warn!("Bridge queue full, dropping oldest message");
728                                }
729                            }
730                        }
731                        Err(e) => {
732                            error!("Failed to receive messages for bridge {}: {}", bridge_id, e);
733                        }
734                    }
735
736                    // Process queued messages in batches
737                    let batch_size = config.batch_size.min(message_queue.len());
738                    if batch_size > 0 {
739                        let batch: Vec<_> = message_queue.drain(..batch_size).collect();
740
741                        for message in batch {
742                            let start_time = Instant::now();
743
744                            match MessageBridgeManager::process_message(
745                                &bridge,
746                                &message,
747                                &transformers,
748                                &router,
749                            )
750                            .await
751                            {
752                                Ok(_) => {
753                                    let duration = start_time.elapsed();
754
755                                    // Update bridge statistics
756                                    MessageBridgeManager::update_bridge_stats(
757                                        &bridges, &bridge_id, true, duration,
758                                    )
759                                    .await;
760                                    stats.write().await.total_messages += 1;
761
762                                    let _ =
763                                        event_notifier.send(BridgeNotification::MessageProcessed {
764                                            bridge_id: bridge_id.clone(),
765                                            message_id: message.id.clone(),
766                                            duration,
767                                        });
768                                }
769                                Err(e) => {
770                                    let duration = start_time.elapsed();
771
772                                    error!(
773                                        "Failed to process message {} in bridge {}: {}",
774                                        message.id, bridge_id, e
775                                    );
776
777                                    // Update bridge statistics
778                                    MessageBridgeManager::update_bridge_stats(
779                                        &bridges, &bridge_id, false, duration,
780                                    )
781                                    .await;
782                                    stats.write().await.failed_messages += 1;
783
784                                    let _ =
785                                        event_notifier.send(BridgeNotification::MessageFailed {
786                                            bridge_id: bridge_id.clone(),
787                                            message_id: message.id.clone(),
788                                            error: e.to_string(),
789                                        });
790
791                                    // Send to dead letter queue if enabled
792                                    if config.enable_dlq {
793                                        // This would implement DLQ logic
794                                        warn!("Message sent to dead letter queue: {}", message.id);
795                                    }
796                                }
797                            }
798                        }
799                    }
800                } else {
801                    // Bridge is not active, exit loop
802                    break;
803                }
804            }
805        });
806
807        Ok(())
808    }
809
810    /// Receive messages from external system
811    async fn receive_messages(
812        source: &ExternalSystemConfig,
813        config: &BridgeConfig,
814    ) -> Result<Vec<ExternalMessage>> {
815        match &source.system_type {
816            ExternalSystemType::Kafka {
817                brokers,
818                topics,
819                consumer_group,
820            } => Self::receive_kafka_messages(brokers, topics, consumer_group, config).await,
821            ExternalSystemType::RabbitMQ {
822                url,
823                exchange,
824                routing_key,
825                queue,
826            } => Self::receive_rabbitmq_messages(url, exchange, routing_key, queue, config).await,
827            ExternalSystemType::RedisPubSub { url, channels } => {
828                Self::receive_redis_messages(url, channels, config).await
829            }
830            ExternalSystemType::HttpRest {
831                base_url,
832                endpoints,
833                headers,
834            } => Self::receive_http_messages(base_url, endpoints, headers, config).await,
835            ExternalSystemType::FileSystem {
836                directory,
837                pattern,
838                watch_mode,
839            } => Self::receive_file_messages(directory, pattern, *watch_mode, config).await,
840            other => Err(anyhow!(
841                "Message receiving is not implemented for system type {:?}",
842                other
843            )),
844        }
845    }
846
847    /// Receive messages from Kafka
848    async fn receive_kafka_messages(
849        _brokers: &[String],
850        _topics: &[String],
851        _consumer_group: &Option<String>,
852        _config: &BridgeConfig,
853    ) -> Result<Vec<ExternalMessage>> {
854        Err(anyhow!(
855            "Kafka bridge consumer transport is not implemented"
856        ))
857    }
858
859    /// Receive messages from RabbitMQ
860    async fn receive_rabbitmq_messages(
861        _url: &str,
862        _exchange: &str,
863        _routing_key: &str,
864        _queue: &Option<String>,
865        _config: &BridgeConfig,
866    ) -> Result<Vec<ExternalMessage>> {
867        Err(anyhow!(
868            "RabbitMQ bridge consumer transport is not implemented"
869        ))
870    }
871
872    /// Receive messages from Redis
873    async fn receive_redis_messages(
874        _url: &str,
875        _channels: &[String],
876        _config: &BridgeConfig,
877    ) -> Result<Vec<ExternalMessage>> {
878        Err(anyhow!(
879            "Redis Pub/Sub bridge consumer transport is not implemented"
880        ))
881    }
882
883    /// Receive messages from HTTP endpoints
884    async fn receive_http_messages(
885        _base_url: &str,
886        _endpoints: &HashMap<String, String>,
887        _headers: &HashMap<String, String>,
888        _config: &BridgeConfig,
889    ) -> Result<Vec<ExternalMessage>> {
890        Err(anyhow!(
891            "HTTP REST bridge consumer transport is not implemented"
892        ))
893    }
894
895    /// Receive messages from the file system.
896    ///
897    /// Reads files under `directory` whose file name matches `pattern` (a simple
898    /// glob supporting a single leading/trailing `*`, or `*` for everything).
899    /// Each matched file becomes one [`ExternalMessage`] whose payload is the
900    /// file contents; the file is then deleted so it is consumed exactly once.
901    async fn receive_file_messages(
902        directory: &str,
903        pattern: &str,
904        _watch_mode: bool,
905        config: &BridgeConfig,
906    ) -> Result<Vec<ExternalMessage>> {
907        let mut messages = Vec::new();
908
909        let mut read_dir = match tokio::fs::read_dir(directory).await {
910            Ok(read_dir) => read_dir,
911            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(messages),
912            Err(e) => {
913                return Err(anyhow!(
914                    "Failed to read bridge source directory {}: {}",
915                    directory,
916                    e
917                ))
918            }
919        };
920
921        while let Some(entry) = read_dir.next_entry().await? {
922            if messages.len() >= config.max_queue_size {
923                break;
924            }
925            let path = entry.path();
926            if !path.is_file() {
927                continue;
928            }
929            let file_name = match path.file_name().and_then(|n| n.to_str()) {
930                Some(name) => name.to_string(),
931                None => continue,
932            };
933            if !Self::file_name_matches(&file_name, pattern) {
934                continue;
935            }
936
937            let payload = tokio::fs::read(&path).await?;
938            let mut metadata = HashMap::new();
939            metadata.insert("file_name".to_string(), file_name.clone());
940            metadata.insert("directory".to_string(), directory.to_string());
941
942            messages.push(ExternalMessage {
943                id: Uuid::new_v4().to_string(),
944                headers: HashMap::new(),
945                payload,
946                format: MessageFormat::Binary,
947                timestamp: chrono::Utc::now(),
948                source: format!("file://{directory}/{file_name}"),
949                metadata,
950            });
951
952            // Consume the file so it is not re-read on the next poll.
953            tokio::fs::remove_file(&path).await?;
954        }
955
956        Ok(messages)
957    }
958
959    /// Match a file name against a simple glob pattern.
960    ///
961    /// Supports `*` (match everything), `*.ext` (suffix), `prefix*` (prefix),
962    /// and exact names. This intentionally covers the common cases without a
963    /// regex dependency.
964    fn file_name_matches(file_name: &str, pattern: &str) -> bool {
965        if pattern == "*" || pattern.is_empty() {
966            return true;
967        }
968        match (pattern.strip_prefix('*'), pattern.strip_suffix('*')) {
969            (Some(suffix), _) if !pattern.ends_with('*') => file_name.ends_with(suffix),
970            (_, Some(prefix)) if !pattern.starts_with('*') => file_name.starts_with(prefix),
971            _ => file_name == pattern,
972        }
973    }
974
975    /// Process a message through the bridge
976    async fn process_message(
977        bridge: &MessageBridge,
978        message: &ExternalMessage,
979        transformers: &Arc<RwLock<HashMap<String, Box<dyn MessageTransformer + Send + Sync>>>>,
980        router: &Arc<RoutingEngine>,
981    ) -> Result<()> {
982        // Apply routing rules
983        let action = router
984            .evaluate_rules(&bridge.routing_rules, message)
985            .await?;
986
987        match action {
988            RuleAction::Drop => {
989                debug!("Message dropped by routing rule: {}", message.id);
990                return Ok(());
991            }
992            RuleAction::Forward => {
993                // Continue with normal processing
994            }
995            RuleAction::Transform { transformer } => {
996                // Apply specific transformer
997                let transformed = {
998                    let transformers_guard = transformers.read().await;
999                    let transformer = transformers_guard
1000                        .get(&transformer)
1001                        .ok_or_else(|| anyhow!("Transformer not found: {}", transformer))?;
1002                    transformer.transform(message)?
1003                };
1004
1005                return Self::send_message(&bridge.target, &transformed).await;
1006            }
1007            _ => {
1008                // Handle other actions
1009                warn!("Routing action not implemented: {:?}", action);
1010            }
1011        }
1012
1013        // Apply default transformation
1014        let transformed = {
1015            let transformers_guard = transformers.read().await;
1016            let transformer = transformers_guard
1017                .get(&bridge.transformer)
1018                .ok_or_else(|| anyhow!("Transformer not found: {}", bridge.transformer))?;
1019            transformer.transform(message)?
1020        };
1021
1022        // Send to target
1023        Self::send_message(&bridge.target, &transformed).await
1024    }
1025
1026    /// Send message to external system
1027    async fn send_message(target: &ExternalSystemConfig, message: &ExternalMessage) -> Result<()> {
1028        match &target.system_type {
1029            ExternalSystemType::Kafka {
1030                brokers, topics, ..
1031            } => Self::send_kafka_message(brokers, topics, message).await,
1032            ExternalSystemType::RabbitMQ {
1033                url,
1034                exchange,
1035                routing_key,
1036                ..
1037            } => Self::send_rabbitmq_message(url, exchange, routing_key, message).await,
1038            ExternalSystemType::RedisPubSub { url, channels } => {
1039                Self::send_redis_message(url, channels, message).await
1040            }
1041            ExternalSystemType::HttpRest {
1042                base_url,
1043                endpoints,
1044                headers,
1045            } => Self::send_http_message(base_url, endpoints, headers, message).await,
1046            ExternalSystemType::FileSystem { directory, .. } => {
1047                Self::send_file_message(directory, message).await
1048            }
1049            other => Err(anyhow!(
1050                "Message sending is not implemented for system type {:?}",
1051                other
1052            )),
1053        }
1054    }
1055
1056    /// Send message to Kafka
1057    async fn send_kafka_message(
1058        _brokers: &[String],
1059        _topics: &[String],
1060        _message: &ExternalMessage,
1061    ) -> Result<()> {
1062        Err(anyhow!(
1063            "Kafka bridge producer transport is not implemented"
1064        ))
1065    }
1066
1067    /// Send message to RabbitMQ
1068    async fn send_rabbitmq_message(
1069        _url: &str,
1070        _exchange: &str,
1071        _routing_key: &str,
1072        _message: &ExternalMessage,
1073    ) -> Result<()> {
1074        Err(anyhow!(
1075            "RabbitMQ bridge producer transport is not implemented"
1076        ))
1077    }
1078
1079    /// Send message to Redis
1080    async fn send_redis_message(
1081        _url: &str,
1082        _channels: &[String],
1083        _message: &ExternalMessage,
1084    ) -> Result<()> {
1085        Err(anyhow!(
1086            "Redis Pub/Sub bridge producer transport is not implemented"
1087        ))
1088    }
1089
1090    /// Send message via HTTP
1091    async fn send_http_message(
1092        _base_url: &str,
1093        _endpoints: &HashMap<String, String>,
1094        _headers: &HashMap<String, String>,
1095        _message: &ExternalMessage,
1096    ) -> Result<()> {
1097        Err(anyhow!(
1098            "HTTP REST bridge producer transport is not implemented"
1099        ))
1100    }
1101
1102    /// Send a message to the file system by writing its payload to a new file in
1103    /// `directory` (named after the message id).
1104    async fn send_file_message(directory: &str, message: &ExternalMessage) -> Result<()> {
1105        tokio::fs::create_dir_all(directory).await?;
1106        let file_path = std::path::Path::new(directory).join(format!("{}.msg", message.id));
1107        tokio::fs::write(&file_path, &message.payload).await?;
1108        debug!("Wrote bridge message {} to {:?}", message.id, file_path);
1109        Ok(())
1110    }
1111
1112    /// Update bridge statistics
1113    async fn update_bridge_stats(
1114        bridges: &Arc<RwLock<HashMap<String, MessageBridge>>>,
1115        bridge_id: &str,
1116        success: bool,
1117        duration: Duration,
1118    ) {
1119        let mut bridges_guard = bridges.write().await;
1120        if let Some(bridge) = bridges_guard.get_mut(bridge_id) {
1121            bridge.last_activity = Some(Instant::now());
1122
1123            if success {
1124                bridge.stats.messages_sent += 1;
1125            } else {
1126                bridge.stats.messages_failed += 1;
1127            }
1128
1129            // Update average processing time
1130            let total_messages = bridge.stats.messages_sent + bridge.stats.messages_failed;
1131            let avg_nanos = bridge.stats.avg_processing_time.as_nanos() as u64;
1132            let duration_nanos = duration.as_nanos() as u64;
1133            if let Some(new_avg_nanos) =
1134                (avg_nanos * (total_messages - 1) + duration_nanos).checked_div(total_messages)
1135            {
1136                bridge.stats.avg_processing_time = Duration::from_nanos(new_avg_nanos);
1137            }
1138        }
1139    }
1140
1141    /// Get bridge statistics
1142    pub async fn get_bridge_stats(&self, bridge_id: &str) -> Result<BridgeStatistics> {
1143        let bridges = self.bridges.read().await;
1144        let bridge = bridges
1145            .get(bridge_id)
1146            .ok_or_else(|| anyhow!("Bridge not found"))?;
1147
1148        Ok(bridge.stats.clone())
1149    }
1150
1151    /// Get manager statistics
1152    pub async fn get_stats(&self) -> BridgeStats {
1153        self.stats.read().await.clone()
1154    }
1155
1156    /// List all bridges
1157    pub async fn list_bridges(&self) -> Vec<BridgeInfo> {
1158        let bridges = self.bridges.read().await;
1159        bridges
1160            .values()
1161            .map(|b| BridgeInfo {
1162                id: b.id.clone(),
1163                bridge_type: b.bridge_type.clone(),
1164                status: format!("{:?}", b.status),
1165                created_at: b.created_at.elapsed(),
1166                last_activity: b.last_activity.map(|t| t.elapsed()),
1167                messages_processed: b.stats.messages_sent + b.stats.messages_failed,
1168                success_rate: if b.stats.messages_sent + b.stats.messages_failed > 0 {
1169                    b.stats.messages_sent as f64
1170                        / (b.stats.messages_sent + b.stats.messages_failed) as f64
1171                } else {
1172                    0.0
1173                },
1174            })
1175            .collect()
1176    }
1177
1178    /// Subscribe to bridge notifications
1179    pub fn subscribe(&self) -> broadcast::Receiver<BridgeNotification> {
1180        self.event_notifier.subscribe()
1181    }
1182}
1183
1184/// Bridge information
1185#[derive(Debug, Clone, Serialize, Deserialize)]
1186pub struct BridgeInfo {
1187    pub id: String,
1188    pub bridge_type: BridgeType,
1189    pub status: String,
1190    pub created_at: Duration,
1191    pub last_activity: Option<Duration>,
1192    pub messages_processed: u64,
1193    pub success_rate: f64,
1194}
1195
1196impl RoutingEngine {
1197    /// Create a new routing engine
1198    fn new() -> Self {
1199        Self {
1200            _global_rules: Arc::new(RwLock::new(Vec::new())),
1201            _rule_cache: Arc::new(RwLock::new(HashMap::new())),
1202        }
1203    }
1204
1205    /// Evaluate routing rules for a message
1206    async fn evaluate_rules(
1207        &self,
1208        rules: &[RoutingRule],
1209        message: &ExternalMessage,
1210    ) -> Result<RuleAction> {
1211        // Sort rules by priority
1212        let mut sorted_rules = rules.to_vec();
1213        sorted_rules.sort_by_key(|r| r.priority);
1214
1215        // Evaluate rules in priority order
1216        for rule in sorted_rules.iter().filter(|r| r.enabled) {
1217            if self.evaluate_condition(&rule.condition, message).await? {
1218                return Ok(rule.action.clone());
1219            }
1220        }
1221
1222        // Default action is forward
1223        Ok(RuleAction::Forward)
1224    }
1225
1226    /// Evaluate a rule condition
1227    #[allow(clippy::only_used_in_recursion)]
1228    fn evaluate_condition<'a>(
1229        &'a self,
1230        condition: &'a RuleCondition,
1231        message: &'a ExternalMessage,
1232    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<bool>> + Send + 'a>> {
1233        Box::pin(async move {
1234            match condition {
1235                RuleCondition::Always => Ok(true),
1236                RuleCondition::EventType { types } => {
1237                    let unknown = "unknown".to_string();
1238                    let event_type = message
1239                        .headers
1240                        .get("event_type")
1241                        .or_else(|| message.metadata.get("event_type"))
1242                        .unwrap_or(&unknown);
1243                    Ok(types.contains(event_type))
1244                }
1245                RuleCondition::Graph { patterns } => {
1246                    let graph = message
1247                        .headers
1248                        .get("graph")
1249                        .or_else(|| message.metadata.get("graph"));
1250                    if let Some(g) = graph {
1251                        Ok(patterns.iter().any(|p| g.contains(p)))
1252                    } else {
1253                        Ok(false)
1254                    }
1255                }
1256                RuleCondition::SubjectPattern { regex } => {
1257                    let subject = message
1258                        .headers
1259                        .get("subject")
1260                        .or_else(|| message.metadata.get("subject"));
1261                    if let Some(s) = subject {
1262                        let re = regex::Regex::new(regex)
1263                            .map_err(|e| anyhow!("Invalid regex: {}", e))?;
1264                        Ok(re.is_match(s))
1265                    } else {
1266                        Ok(false)
1267                    }
1268                }
1269                RuleCondition::Predicate { predicates } => {
1270                    let predicate = message
1271                        .headers
1272                        .get("predicate")
1273                        .or_else(|| message.metadata.get("predicate"));
1274                    if let Some(p) = predicate {
1275                        Ok(predicates.contains(p))
1276                    } else {
1277                        Ok(false)
1278                    }
1279                }
1280                RuleCondition::Expression { expr } => {
1281                    // This would implement expression evaluation
1282                    warn!("Expression evaluation not implemented: {}", expr);
1283                    Ok(false)
1284                }
1285                RuleCondition::Composite {
1286                    operator,
1287                    conditions,
1288                } => match operator {
1289                    LogicalOperator::And => {
1290                        for cond in conditions {
1291                            if !self.evaluate_condition(cond, message).await? {
1292                                return Ok(false);
1293                            }
1294                        }
1295                        Ok(true)
1296                    }
1297                    LogicalOperator::Or => {
1298                        for cond in conditions {
1299                            if self.evaluate_condition(cond, message).await? {
1300                                return Ok(true);
1301                            }
1302                        }
1303                        Ok(false)
1304                    }
1305                    LogicalOperator::Not => {
1306                        if conditions.len() != 1 {
1307                            return Err(anyhow!("NOT operator requires exactly one condition"));
1308                        }
1309                        Ok(!self.evaluate_condition(&conditions[0], message).await?)
1310                    }
1311                },
1312            }
1313        })
1314    }
1315}
1316
1317/// JSON message transformer
1318pub struct JsonTransformer;
1319
1320impl MessageTransformer for JsonTransformer {
1321    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage> {
1322        // This would implement JSON transformation logic
1323        Ok(message.clone())
1324    }
1325
1326    fn name(&self) -> &str {
1327        "json"
1328    }
1329
1330    fn supported_formats(&self) -> (MessageFormat, MessageFormat) {
1331        (MessageFormat::Json, MessageFormat::Json)
1332    }
1333}
1334
1335/// RDF to JSON transformer
1336pub struct RdfToJsonTransformer;
1337
1338impl MessageTransformer for RdfToJsonTransformer {
1339    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage> {
1340        // This would implement RDF to JSON transformation
1341        let mut transformed = message.clone();
1342        transformed.format = MessageFormat::Json;
1343
1344        // Transform payload from RDF to JSON
1345        // For now, just pass through
1346
1347        Ok(transformed)
1348    }
1349
1350    fn name(&self) -> &str {
1351        "rdf-to-json"
1352    }
1353
1354    fn supported_formats(&self) -> (MessageFormat, MessageFormat) {
1355        (MessageFormat::Text, MessageFormat::Json) // Assuming RDF as text
1356    }
1357}
1358
1359#[cfg(test)]
1360mod tests {
1361    use super::*;
1362
1363    #[tokio::test]
1364    async fn test_bridge_creation() {
1365        let manager = MessageBridgeManager::new().await.unwrap();
1366
1367        let source = ExternalSystemConfig {
1368            system_type: ExternalSystemType::Kafka {
1369                brokers: vec!["localhost:9092".to_string()],
1370                topics: vec!["source-topic".to_string()],
1371                consumer_group: Some("test-group".to_string()),
1372            },
1373            connection: ConnectionConfig {
1374                timeout: Duration::from_secs(30),
1375                keep_alive: Duration::from_secs(60),
1376                retry: RetryConfig {
1377                    max_attempts: 3,
1378                    initial_delay: Duration::from_millis(100),
1379                    max_delay: Duration::from_secs(10),
1380                    exponential_backoff: true,
1381                },
1382                tls: None,
1383            },
1384            format: FormatConfig {
1385                format: MessageFormat::Json,
1386                encoding: "utf-8".to_string(),
1387                compression: None,
1388                schema_validation: false,
1389            },
1390            security: SecurityConfig {
1391                auth: AuthenticationMethod::None,
1392                encryption: EncryptionConfig {
1393                    enabled: false,
1394                    algorithm: None,
1395                    key_id: None,
1396                },
1397                access_control: AccessControlConfig {
1398                    read_permissions: vec![],
1399                    write_permissions: vec![],
1400                    admin_permissions: vec![],
1401                },
1402            },
1403        };
1404
1405        let target = source.clone(); // Same config for simplicity
1406
1407        // Register transformer
1408        manager
1409            .register_transformer(Box::new(JsonTransformer))
1410            .await;
1411
1412        let bridge_id = manager
1413            .create_bridge(
1414                BridgeType::SourceToTarget,
1415                source,
1416                target,
1417                "json".to_string(),
1418                vec![],
1419                BridgeConfig::default(),
1420            )
1421            .await
1422            .unwrap();
1423
1424        assert!(!bridge_id.is_empty());
1425
1426        let bridges = manager.list_bridges().await;
1427        assert_eq!(bridges.len(), 1);
1428        assert_eq!(bridges[0].id, bridge_id);
1429    }
1430
1431    #[tokio::test]
1432    async fn test_routing_rules() {
1433        let engine = RoutingEngine::new();
1434
1435        let rule = RoutingRule {
1436            name: "test-rule".to_string(),
1437            condition: RuleCondition::EventType {
1438                types: vec!["triple_added".to_string()],
1439            },
1440            action: RuleAction::Forward,
1441            priority: 1,
1442            enabled: true,
1443        };
1444
1445        let mut message = ExternalMessage {
1446            id: "test".to_string(),
1447            headers: HashMap::new(),
1448            payload: vec![],
1449            format: MessageFormat::Json,
1450            timestamp: chrono::Utc::now(),
1451            source: "test".to_string(),
1452            metadata: HashMap::new(),
1453        };
1454
1455        message
1456            .headers
1457            .insert("event_type".to_string(), "triple_added".to_string());
1458
1459        let action = engine.evaluate_rules(&[rule], &message).await.unwrap();
1460        assert!(matches!(action, RuleAction::Forward));
1461    }
1462
1463    #[test]
1464    fn regression_file_name_matches_glob() {
1465        assert!(MessageBridgeManager::file_name_matches("a.json", "*"));
1466        assert!(MessageBridgeManager::file_name_matches("a.json", "*.json"));
1467        assert!(!MessageBridgeManager::file_name_matches("a.txt", "*.json"));
1468        assert!(MessageBridgeManager::file_name_matches("data_1", "data_*"));
1469        assert!(!MessageBridgeManager::file_name_matches("other", "data_*"));
1470        assert!(MessageBridgeManager::file_name_matches("exact", "exact"));
1471    }
1472
1473    #[tokio::test]
1474    async fn regression_file_transport_roundtrip() {
1475        let dir = std::env::temp_dir().join(format!("oxirs-bridge-{}", Uuid::new_v4()));
1476        tokio::fs::create_dir_all(&dir).await.unwrap();
1477        let dir_str = dir.to_string_lossy().to_string();
1478
1479        // Write a message via the real file sender.
1480        let message = ExternalMessage {
1481            id: "msg-1".to_string(),
1482            headers: HashMap::new(),
1483            payload: b"hello-bridge".to_vec(),
1484            format: MessageFormat::Binary,
1485            timestamp: chrono::Utc::now(),
1486            source: "test".to_string(),
1487            metadata: HashMap::new(),
1488        };
1489        MessageBridgeManager::send_file_message(&dir_str, &message)
1490            .await
1491            .unwrap();
1492
1493        // Receiving must actually read the file back (not return empty).
1494        let received = MessageBridgeManager::receive_file_messages(
1495            &dir_str,
1496            "*",
1497            false,
1498            &BridgeConfig::default(),
1499        )
1500        .await
1501        .unwrap();
1502        assert_eq!(received.len(), 1);
1503        assert_eq!(received[0].payload, b"hello-bridge");
1504
1505        // File is consumed: a second receive returns nothing.
1506        let again = MessageBridgeManager::receive_file_messages(
1507            &dir_str,
1508            "*",
1509            false,
1510            &BridgeConfig::default(),
1511        )
1512        .await
1513        .unwrap();
1514        assert!(again.is_empty());
1515
1516        let _ = tokio::fs::remove_dir_all(&dir).await;
1517    }
1518
1519    #[tokio::test]
1520    async fn regression_start_bridge_rejects_unimplemented_transport() {
1521        let manager = MessageBridgeManager::new().await.unwrap();
1522        manager
1523            .register_transformer(Box::new(JsonTransformer))
1524            .await;
1525
1526        let kafka = ExternalSystemConfig {
1527            system_type: ExternalSystemType::Kafka {
1528                brokers: vec!["localhost:9092".to_string()],
1529                topics: vec!["t".to_string()],
1530                consumer_group: None,
1531            },
1532            connection: ConnectionConfig {
1533                timeout: Duration::from_secs(1),
1534                keep_alive: Duration::from_secs(1),
1535                retry: RetryConfig {
1536                    max_attempts: 1,
1537                    initial_delay: Duration::from_millis(1),
1538                    max_delay: Duration::from_millis(1),
1539                    exponential_backoff: false,
1540                },
1541                tls: None,
1542            },
1543            format: FormatConfig {
1544                format: MessageFormat::Json,
1545                encoding: "utf-8".to_string(),
1546                compression: None,
1547                schema_validation: false,
1548            },
1549            security: SecurityConfig {
1550                auth: AuthenticationMethod::None,
1551                encryption: EncryptionConfig {
1552                    enabled: false,
1553                    algorithm: None,
1554                    key_id: None,
1555                },
1556                access_control: AccessControlConfig {
1557                    read_permissions: vec![],
1558                    write_permissions: vec![],
1559                    admin_permissions: vec![],
1560                },
1561            },
1562        };
1563
1564        let bridge_id = manager
1565            .create_bridge(
1566                BridgeType::SourceToTarget,
1567                kafka.clone(),
1568                kafka,
1569                "json".to_string(),
1570                vec![],
1571                BridgeConfig::default(),
1572            )
1573            .await
1574            .unwrap();
1575
1576        // Starting a bridge with an unimplemented transport must fail loud.
1577        assert!(manager.start_bridge(&bridge_id).await.is_err());
1578    }
1579}
oxirs_stream/bridge.rs

oxirs_stream/
bridge.rs
