Fix CI failures in clippy, coverage, and windows stream bus build

Author: argahsuknesib

examples/debug_live.rs

@@ -2,7 +2,6 @@ use janus::core::RDFEvent;
 use janus::stream::live_stream_processing::LiveStreamProcessing;
 use std::thread;
 use std::time::Duration;
-use std::time::{SystemTime, UNIX_EPOCH};
 
 fn main() {
     println!("Starting debug_live reproduction...");

examples/http_client_example.rs

@@ -10,6 +10,7 @@
 //!
 //! Usage:
 //!   cargo run --example http_client_example
+#![allow(dead_code)]
 
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;

examples/test_query_pipeline.rs

@@ -40,7 +40,7 @@ WHERE {
     let events = storage.query(0, u64::MAX).expect("Storage query failed");
     println!("Storage has {} events", events.len());
 
-    if events.len() > 0 {
+    if !events.is_empty() {
         let dict = storage.get_dictionary().read().unwrap();
         println!("\nFirst 3 events decoded:");
         for (i, e) in events.iter().take(3).enumerate() {

examples/test_storage_query.rs

@@ -17,7 +17,7 @@ fn main() {
 
     println!("Total events in storage: {}", events.len());
 
-    if events.len() > 0 {
+    if !events.is_empty() {
         println!("\nFirst 3 events:");
         for (i, event) in events.iter().take(3).enumerate() {
             println!(

examples/test_storage_with_dict.rs

@@ -17,7 +17,7 @@ fn main() {
 
     println!("Total events in storage: {}", events.len());
 
-    if events.len() > 0 {
+    if !events.is_empty() {
         let dict = storage.get_dictionary().read().unwrap();
         println!("\nDecoded first 5 events:");
         for (i, e) in events.iter().take(5).enumerate() {

src/lib.rs

@@ -53,6 +53,25 @@
 #![allow(clippy::doc_markdown)]
 #![allow(clippy::identity_op)]
 #![allow(clippy::needless_update)]
+#![allow(clippy::needless_raw_string_hashes)]
+#![allow(clippy::unreadable_literal)]
+#![allow(clippy::similar_names)]
+#![allow(clippy::redundant_else)]
+#![allow(clippy::type_complexity)]
+#![allow(clippy::redundant_pattern_matching)]
+#![allow(clippy::manual_string_new)]
+#![allow(clippy::ignored_unit_patterns)]
+#![allow(clippy::unnecessary_wraps)]
+#![allow(clippy::if_same_then_else)]
+#![allow(clippy::too_many_lines)]
+#![allow(clippy::match_wildcard_for_single_variants)]
+#![allow(clippy::module_inception)]
+#![allow(clippy::cast_precision_loss)]
+#![allow(clippy::single_char_pattern)]
+#![allow(clippy::unnecessary_debug_formatting)]
+#![allow(clippy::elidable_lifetime_names)]
+#![allow(clippy::cast_sign_loss)]
+#![allow(clippy::map_unwrap_or)]
 #![allow(missing_docs)]
 
 /// Core data structures and types

src/stream_bus/stream_bus.rs

@@ -10,7 +10,9 @@ use crate::core::RDFEvent;
 use crate::parsing::rdf_parser;
 use crate::storage::segmented_storage::StreamingSegmentedStorage;
 use core::str;
+#[cfg(not(windows))]
 use rdkafka::config::ClientConfig;
+#[cfg(not(windows))]
 use rdkafka::producer::{FutureProducer, FutureRecord};
 use rumqttc::{AsyncClient, MqttOptions, QoS};
 use std::fmt::write;
@@ -268,6 +270,7 @@ impl StreamBus {
         }
     }
 
+    #[cfg(not(windows))]
     async fn run_with_kafka(&self) -> Result<(), StreamBusError> {
         let kafka_config =
             self.config.kafka_config.as_ref().ok_or(StreamBusError::ConfigError(
@@ -311,6 +314,13 @@ impl StreamBus {
         .await
     }
 
+    #[cfg(windows)]
+    async fn run_with_kafka(&self) -> Result<(), StreamBusError> {
+        Err(StreamBusError::BrokerError(
+            "Kafka broker mode is not supported on Windows builds".to_string(),
+        ))
+    }
+
     async fn run_with_mqtt(&self) -> Result<(), StreamBusError> {
         let mqtt_config = self
             .config

tests/stream_bus_cli_test.rs

@@ -36,11 +36,30 @@ fn create_test_rdf_file(path: &str, num_events: usize) -> std::io::Result<()> {
 }
 
 fn get_cli_binary() -> String {
-    if Path::new("target/debug/stream_bus_cli").exists() {
-        "target/debug/stream_bus_cli".to_string()
+    if let Ok(path) = std::env::var("CARGO_BIN_EXE_stream_bus_cli") {
+        return path;
+    }
+
+    let bin_name = if cfg!(windows) {
+        "stream_bus_cli.exe"
     } else {
-        "target/release/stream_bus_cli".to_string()
+        "stream_bus_cli"
+    };
+
+    let candidates = [
+        format!("target/debug/{bin_name}"),
+        format!("target/release/{bin_name}"),
+        format!("target/llvm-cov-target/debug/{bin_name}"),
+        format!("target/llvm-cov-target/release/{bin_name}"),
+    ];
+
+    for candidate in candidates {
+        if Path::new(&candidate).exists() {
+            return candidate;
+        }
     }
+
+    panic!("Could not find stream_bus_cli binary in expected target locations");
 }
 
 #[test]

tests/stream_bus_test.rs

@@ -66,7 +66,7 @@ fn test_parse_ntriples_basic() {
     };
 
     let storage = create_test_storage(&test_dir).unwrap();
-    let bus = StreamBus::new(config, storage);
+    let _bus = StreamBus::new(config, storage);
 
     let line = "<http://example.org/sensor1> <http://example.org/temperature> \"23.5\" <http://example.org/graph1> .";
     let event = rdf_parser::parse_rdf_line(line, true);
@@ -97,7 +97,7 @@ fn test_parse_ntriples_without_graph() {
     };
 
     let storage = create_test_storage(&test_dir).unwrap();
-    let bus = StreamBus::new(config, storage);
+    let _bus = StreamBus::new(config, storage);
 
     let line = "<http://example.org/alice> <http://example.org/knows> <http://example.org/bob> .";
     let event = rdf_parser::parse_rdf_line(line, true);
@@ -128,7 +128,7 @@ fn test_parse_invalid_rdf_line() {
     };
 
     let storage = create_test_storage(&test_dir).unwrap();
-    let bus = StreamBus::new(config, storage);
+    let _bus = StreamBus::new(config, storage);
 
     let invalid_line = "<http://example.org/subject> <http://example.org/predicate>";
     let result = rdf_parser::parse_rdf_line(invalid_line, true);
@@ -380,7 +380,7 @@ fn test_timestamp_parsing() {
     };
 
     let storage = create_test_storage(&test_dir).unwrap();
-    let bus_with_ts = StreamBus::new(config_with_timestamps, Arc::clone(&storage));
+    let _bus_with_ts = StreamBus::new(config_with_timestamps, Arc::clone(&storage));
 
     let line = "<http://example.org/sensor1> <http://example.org/temperature> \"23.5\" <http://example.org/graph1> .";
     let event = rdf_parser::parse_rdf_line(line, true).unwrap();
@@ -397,7 +397,7 @@ fn test_timestamp_parsing() {
         mqtt_config: None,
     };
 
-    let bus_without_ts = StreamBus::new(config_without_timestamps, storage);
+    let _bus_without_ts = StreamBus::new(config_without_timestamps, storage);
 
     let line_with_ts = "1234567890 <http://example.org/sensor1> <http://example.org/ts> \"value\" <http://example.org/graph1> .";
     let event = rdf_parser::parse_rdf_line(line_with_ts, false).unwrap();