Update deps

Cargo.toml

@@ -39,7 +39,6 @@ cassandra-protocol = "3.0"
 tracing = "0.1.15"
 tracing-subscriber = { version = "0.3.1", features = ["env-filter", "json"] }
 tracing-appender = "0.2.0"
-nix = { version = "0.27.0", features = ["signal"]}
 serde_json = "1.0"
 rcgen = "0.12.0"
 subprocess = "0.2.7"
@@ -56,7 +55,7 @@ clap = { version = "4.0.4", features = ["cargo", "derive"] }
 async-trait = "0.1.30"
 typetag = "0.2.5"
 aws-throwaway = { version = "0.6.0", default-features = false }
-tokio-bin-process = "0.4.0"
+tokio-bin-process = "0.5.0"
 ordered-float = { version = "4.0.0", features = ["serde"] }
 hyper = { version = "0.14.14", features = ["server", "tcp", "http1"] }
 shell-quote = { default-features = false, version = "0.5.0" }

shotover-proxy/Cargo.toml

@@ -28,7 +28,6 @@ serde.workspace = true
 csv.workspace = true
 uuid.workspace = true
 itertools.workspace = true
-nix.workspace = true
 cdrs-tokio.workspace = true
 redis-protocol.workspace = true
 tokio-util.workspace = true

shotover-proxy/benches/windsock/profilers.rs

@@ -60,7 +60,7 @@ impl ProfilerRunner {
                     Samply::run(
                         self.bench_name.clone(),
                         self.results_path.clone(),
-                        shotover.child().id().unwrap(),
+                        shotover.pid(),
                     )
                     .await,
                 )

shotover-proxy/benches/windsock/profilers/samply.rs

@@ -9,7 +9,7 @@ pub struct Samply {
 }
 
 impl Samply {
-    pub async fn run(bench_name: String, results_path: PathBuf, pid: u32) -> Samply {
+    pub async fn run(bench_name: String, results_path: PathBuf, pid: i32) -> Samply {
         run_command(
             "cargo",
             &[

shotover-proxy/tests/runner/runner_int_tests.rs

@@ -17,7 +17,7 @@ async fn test_shotover_responds_sigterm() {
         let shotover_process = shotover_process("tests/test-configs/null-redis/topology.yaml")
             .start()
             .await;
-        shotover_process.signal(nix::sys::signal::Signal::SIGTERM);
+        shotover_process.send_sigterm();
 
         let events = shotover_process.consume_remaining_events(&[]).await;
         events.assert_contains(
@@ -34,7 +34,7 @@ async fn test_shotover_responds_sigint() {
     let shotover_process = shotover_process("tests/test-configs/null-redis/topology.yaml")
         .start()
         .await;
-    shotover_process.signal(nix::sys::signal::Signal::SIGINT);
+    shotover_process.send_sigint();
 
     let events = shotover_process.consume_remaining_events(&[]).await;
     events.assert_contains(

shotover/Cargo.toml

@@ -38,6 +38,7 @@ redis = [
     "dep:redis-protocol",
     "dep:csv",
     "dep:crc16",
+    "dep:bytes-utils"
 ]
 opensearch = [
     "dep:atoi",
@@ -47,12 +48,12 @@ opensearch = [
 default = ["cassandra", "redis", "kafka", "opensearch"]
 
 [dependencies]
-atomic_enum = "0.2.0"
+atomic_enum = "0.3.0"
 pretty-hex = "0.4.0"
 tokio-stream = "0.1.2"
-bytes-utils = "0.1.1"
+bytes-utils = { version = "0.1.1", optional = true }
 derivative = "2.1.1"
-cached = { version = "0.48", features = ["async"], optional = true }
+cached = { version = "0.49", features = ["async"], optional = true }
 governor = { version = "0.6", default-features = false, features = ["std", "jitter", "quanta"] }
 nonzero_ext = "0.3.0"
 version-compare = { version = "0.1", optional = true }
@@ -110,7 +111,7 @@ aws-config = { version = "1.0.0", optional = true }
 aws-sdk-kms = { version = "1.1.0", optional = true }
 chacha20poly1305 = { version = "0.10.0", features = ["std"], optional = true }
 generic-array = { version = "0.14", features = ["serde"], optional = true }
-kafka-protocol = { version = "0.8.0", optional = true }
+kafka-protocol = { version = "0.9.0", optional = true }
 rustls = { version = "0.22.0" }
 tokio-rustls = "0.25"
 rustls-pemfile = "2.0.0"

shotover/src/frame/kafka.rs

@@ -43,9 +43,11 @@ use kafka_protocol::messages::{
     UpdateMetadataResponse, VoteRequest, VoteResponse, WriteTxnMarkersRequest,
     WriteTxnMarkersResponse,
 };
-use kafka_protocol::protocol::{Decodable, Encodable, HeaderVersion, StrBytes};
+use kafka_protocol::protocol::{Decodable, Encodable, HeaderVersion};
 use std::fmt::{Display, Formatter, Result as FmtResult};
 
+pub use kafka_protocol::protocol::StrBytes;
+
 #[derive(Debug, PartialEq, Clone)]
 pub enum KafkaFrame {
     Request {
@@ -872,15 +874,3 @@ fn encode<T: Encodable>(encodable: T, bytes: &mut BytesMut, version: i16) -> Res
         )
     })
 }
-
-/// This function is a helper to workaround a really degenerate rust compiler case.
-/// The problem is that the string crate defines a TryFrom which collides with the stdlib TryFrom
-/// and then naming the correct TryFrom becomes really annoying.
-pub fn strbytes(str: &str) -> StrBytes {
-    <StrBytes as string::TryFrom<Bytes>>::try_from(Bytes::copy_from_slice(str.as_bytes())).unwrap()
-}
-
-/// Allocationless version of kafka_strbytes
-pub fn strbytes_static(str: &'static str) -> StrBytes {
-    <StrBytes as string::TryFrom<Bytes>>::try_from(Bytes::from(str)).unwrap()
-}

shotover/src/transforms/kafka/sink_cluster/mod.rs

@@ -1,5 +1,5 @@
 use super::common::produce_channel;
-use crate::frame::kafka::{strbytes, KafkaFrame, RequestBody, ResponseBody};
+use crate::frame::kafka::{KafkaFrame, RequestBody, ResponseBody};
 use crate::frame::Frame;
 use crate::message::{Message, Messages};
 use crate::tls::{TlsConnector, TlsConnectorConfig};
@@ -92,7 +92,7 @@ impl KafkaSinkClusterBuilder {
             .map(|node| {
                 let address: SocketAddr = node.parse().unwrap();
                 KafkaAddress {
-                    host: strbytes(&address.ip().to_string()),
+                    host: StrBytes::from_string(address.ip().to_string()),
                     port: address.port() as i32,
                 }
             })

shotover/src/transforms/kafka/sink_cluster/node.rs

@@ -1,5 +1,4 @@
 use crate::codec::{kafka::KafkaCodecBuilder, CodecBuilder, Direction};
-use crate::frame::kafka::strbytes;
 use crate::tcp;
 use crate::tls::TlsConnector;
 use crate::transforms::util::cluster_connection_pool::{spawn_read_write_tasks, Connection};
@@ -23,10 +22,11 @@ impl KafkaAddress {
     pub fn from_str(address: &str) -> Result<Self> {
         let mut address_iter = address.split(':');
         Ok(KafkaAddress {
-            host: strbytes(
+            host: StrBytes::from_string(
                 address_iter
                     .next()
-                    .ok_or_else(|| anyhow!("Address must include ':' seperator"))?,
+                    .ok_or_else(|| anyhow!("Address must include ':' seperator"))?
+                    .to_owned(),
             ),
             port: address_iter
                 .next()