operate on line bytes to allow handling of plaintext lines

estuary · Apr 14, 2022 · 69bccea · 69bccea
1 parent beeb2d3
commit 69bccea
Show file tree

Hide file tree

Showing 5 changed files with 160 additions and 117 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/connector_proxy/Cargo.toml b/crates/connector_proxy/Cargo.toml
@@ -33,5 +33,7 @@ tempfile="*"
 thiserror = "*"
 tokio = { version = "1.15.0", features = ["full"] }
 tokio-util = { version = "*", features = ["io"] }
+tokio-stream = { version = "*", features = ["io-util"] }
+bytelines = "*"
 tracing="*"
 validator = { version = "*", features = ["derive"] }
diff --git a/crates/connector_proxy/src/interceptors/network_tunnel_capture_interceptor.rs b/crates/connector_proxy/src/interceptors/network_tunnel_capture_interceptor.rs
@@ -2,14 +2,14 @@ use crate::apis::{FlowCaptureOperation, InterceptorStream};
 use crate::errors::{Error, Must};
 use crate::libs::network_tunnel::NetworkTunnel;
 use crate::libs::protobuf::{decode_message, encode_message};
-use crate::libs::stream::{get_decoded_message, stream_all_bytes};
+use crate::libs::stream::get_decoded_message;
 use futures::{future, stream, StreamExt, TryStreamExt};
 use protocol::capture::{
     ApplyRequest, DiscoverRequest, PullRequest, SpecResponse, ValidateRequest,
 };
 
 use serde_json::value::RawValue;
-use tokio_util::io::StreamReader;
+use tokio_util::io::{ReaderStream, StreamReader};
 
 pub struct NetworkTunnelCaptureInterceptor {}
 
@@ -81,7 +81,7 @@ impl NetworkTunnelCaptureInterceptor {
                 }
 
                 let first = stream::once(future::ready(encode_message(&request)));
-                let rest = stream_all_bytes(reader);
+                let rest = ReaderStream::new(reader);
 
                 // We need to set explicit error type, see https://github.com/rust-lang/rust/issues/63502
                 Ok::<_, std::io::Error>(first.chain(rest))

diff --git a/crates/connector_proxy/src/interceptors/network_tunnel_materialize_interceptor.rs b/crates/connector_proxy/src/interceptors/network_tunnel_materialize_interceptor.rs
@@ -2,13 +2,13 @@ use crate::apis::{FlowMaterializeOperation, InterceptorStream};
 use crate::errors::{Error, Must};
 use crate::libs::network_tunnel::NetworkTunnel;
 use crate::libs::protobuf::{decode_message, encode_message};
-use crate::libs::stream::{get_decoded_message, stream_all_bytes};
+use crate::libs::stream::get_decoded_message;
 
 use futures::{future, stream, StreamExt, TryStreamExt};
 use protocol::materialize::{ApplyRequest, SpecResponse, TransactionRequest, ValidateRequest};
 
 use serde_json::value::RawValue;
-use tokio_util::io::StreamReader;
+use tokio_util::io::{ReaderStream, StreamReader};
 
 pub struct NetworkTunnelMaterializeInterceptor {}
 
@@ -63,7 +63,7 @@ impl NetworkTunnelMaterializeInterceptor {
                     }
                 }
                 let first = stream::once(future::ready(encode_message(&request)));
-                let rest = stream_all_bytes(reader);
+                let rest = ReaderStream::new(reader);
 
                 // We need to set explicit error type, see https://github.com/rust-lang/rust/issues/63502
                 Ok::<_, std::io::Error>(first.chain(rest))