bridge: update edition to 2024

Author: svix-jbrown

.justfile-common-rust

@@ -8,7 +8,7 @@ fmt:
 
 # Run `cargo clippy` for the rfust codebase.
 clippy *args='':
-    cargo clippy --all-targets {{ args }} -- -D warnings
+    cargo clippy --all-targets --all-features {{ args }} -- -D warnings
 
 # Run lints (fmt and clippy)
 lint: fmt clippy && fmt

bridge/.justfile

@@ -12,4 +12,4 @@ run *args='':
 # Invokes `docker compose` with the testing configuration in the 'bridge' directory
 [no-exit-message]
 dc *args='':
-    docker compose -p svix-oss -f {{ HERE / "testing-docker-compose.yml" }} {{ args }}
+    docker compose -p svix-bridge -f {{ HERE / "testing-docker-compose.yml" }} {{ args }}

bridge/Cargo.lock

@@ -1,7 +1,7 @@
 [package]
 name = "svix-bridge-plugin-kafka"
 version.workspace = true
-edition = "2021"
+edition = "2024"
 rust-version = "1.91"
 publish = false
 license.workspace = true

bridge/svix-bridge-plugin-kafka/Cargo.toml

@@ -1,10 +1,10 @@
 use rdkafka::{
-    consumer::StreamConsumer, error::KafkaResult, producer::FutureProducer, ClientConfig,
+    ClientConfig, consumer::StreamConsumer, error::KafkaResult, producer::FutureProducer,
 };
 use serde::Deserialize;
 use svix_bridge_types::{ReceiverOutput, SenderInput, SenderOutputOpts, TransformationConfig};
 
-use crate::{input::KafkaConsumer, KafkaProducer, Result};
+use crate::{KafkaProducer, Result, input::KafkaConsumer};
 
 #[derive(Clone, Deserialize)]
 #[serde(tag = "type")]
@@ -56,10 +56,10 @@ impl KafkaInputOpts {
             .set("enable.auto.commit", "false");
 
         security_protocol.apply(&mut config);
-        if let Some(debug_contexts) = debug_contexts {
-            if !debug_contexts.is_empty() {
-                config.set("debug", debug_contexts);
-            }
+        if let Some(debug_contexts) = debug_contexts
+            && !debug_contexts.is_empty()
+        {
+            config.set("debug", debug_contexts);
         }
 
         config.create()
@@ -106,10 +106,10 @@ impl KafkaOutputOpts {
         config.set("bootstrap.servers", bootstrap_brokers);
 
         security_protocol.apply(&mut config);
-        if let Some(debug_contexts) = debug_contexts {
-            if !debug_contexts.is_empty() {
-                config.set("debug", debug_contexts);
-            }
+        if let Some(debug_contexts) = debug_contexts
+            && !debug_contexts.is_empty()
+        {
+            config.set("debug", debug_contexts);
         }
 
         config.create()

bridge/svix-bridge-plugin-kafka/src/config.rs

@@ -4,19 +4,19 @@ use std::{
 };
 
 use rdkafka::{
+    Message as _,
     consumer::{CommitMode, Consumer as _},
     error::KafkaError,
-    Message as _,
 };
 use svix_bridge_types::{
-    async_trait,
-    svix::api::{MessageCreateOptions, Svix},
     CreateMessageRequest, JsObject, SenderInput, SenderOutputOpts, TransformationConfig,
     TransformerInput, TransformerInputFormat, TransformerJob, TransformerOutput, TransformerTx,
+    async_trait,
+    svix::api::{MessageCreateOptions, Svix},
 };
 use tokio::task::spawn_blocking;
 
-use crate::{config::KafkaInputOpts, Error, Result};
+use crate::{Error, Result, config::KafkaInputOpts};
 
 pub struct KafkaConsumer {
     name: String,

bridge/svix-bridge-plugin-kafka/src/input.rs

@@ -5,8 +5,8 @@ mod output;
 
 pub use self::{
     config::{
-        into_receiver_output, into_sender_input, KafkaInputOpts, KafkaOutputOpts,
-        KafkaSecurityProtocol,
+        KafkaInputOpts, KafkaOutputOpts, KafkaSecurityProtocol, into_receiver_output,
+        into_sender_input,
     },
     error::{Error, Result},
     input::KafkaConsumer,

bridge/svix-bridge-plugin-kafka/src/lib.rs

@@ -3,7 +3,7 @@ use rdkafka::{
     producer::{FutureProducer, FutureRecord},
     util::Timeout,
 };
-use svix_bridge_types::{async_trait, BoxError, ForwardRequest, ReceiverOutput};
+use svix_bridge_types::{BoxError, ForwardRequest, ReceiverOutput, async_trait};
 
 use crate::config::KafkaOutputOpts;
 

bridge/svix-bridge-plugin-kafka/src/output.rs

@@ -5,24 +5,24 @@
 use std::time::Duration;
 
 use rdkafka::{
+    ClientConfig,
     producer::{FutureProducer, FutureRecord},
     util::Timeout,
-    ClientConfig,
 };
 use serde_json::json;
 use svix_bridge_plugin_kafka::{KafkaConsumer, KafkaInputOpts};
 use svix_bridge_types::{
-    svix::api::MessageIn, CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions,
-    SvixSenderOutputOpts, TransformationConfig, TransformerInput, TransformerInputFormat,
-    TransformerJob, TransformerOutput,
+    CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts,
+    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
+    TransformerOutput, svix::api::MessageIn,
 };
 use tracing::info;
 use wiremock::{
-    matchers::{body_partial_json, method},
     Mock, MockServer, ResponseTemplate,
+    matchers::{body_partial_json, method},
 };
 
-use crate::{create_topic, delete_topic, kafka_admin_client, BROKER_HOST};
+use crate::{BROKER_HOST, create_topic, delete_topic, kafka_admin_client};
 
 #[ctor::ctor]
 fn test_setup() {

bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs

@@ -1,14 +1,14 @@
 use std::{sync::Arc, time::Duration};
 
 use rdkafka::{
-    consumer::{Consumer, StreamConsumer},
     ClientConfig, Message,
+    consumer::{Consumer, StreamConsumer},
 };
 use serde_json::json;
 use svix_bridge_plugin_kafka::{KafkaOutputOpts, KafkaProducer};
 use svix_bridge_types::{ForwardRequest, ReceiverOutput as _};
 
-use crate::{create_topic, delete_topic, kafka_admin_client, BROKER_HOST};
+use crate::{BROKER_HOST, create_topic, delete_topic, kafka_admin_client};
 
 /// Time to wait for the consumer to be properly listening.
 const LISTEN_WAIT_TIME: Duration = Duration::from_secs(8);