cleanup

src/sinks/http/batch.rs

@@ -1,10 +1,13 @@
+//! Batch settings for the `http` sink.
+
 use codecs::encoding::Framer;
 use vector_core::{
     event::Event, stream::batcher::limiter::ItemBatchSize, ByteSizeOf, EstimatedJsonEncodedSizeOf,
 };
 
 use crate::codecs::Encoder;
 
+/// Uses the configured encoder to determine batch sizing.
 #[derive(Default)]
 pub(super) struct HttpBatchSizer {
     pub(super) encoder: Encoder<Framer>,

src/sinks/http/config.rs

@@ -35,33 +35,33 @@ const CONTENT_TYPE_JSON: &str = "application/json";
 #[configurable_component(sink("http", "Deliver observability event data to an HTTP server."))]
 #[derive(Clone, Debug)]
 #[serde(deny_unknown_fields)]
-pub struct HttpSinkConfig {
+pub(super) struct HttpSinkConfig {
     /// The full URI to make HTTP requests to.
     ///
     /// This should include the protocol and host, but can also include the port, path, and any other valid part of a URI.
     #[configurable(metadata(docs::examples = "https://10.22.212.22:9000/endpoint"))]
-    pub uri: UriSerde,
+    pub(super) uri: UriSerde,
 
     /// The HTTP method to use when making the request.
     #[serde(default)]
-    pub method: HttpMethod,
+    pub(super) method: HttpMethod,
 
     #[configurable(derived)]
-    pub auth: Option<Auth>,
+    pub(super) auth: Option<Auth>,
 
     /// A list of custom headers to add to each request.
     #[configurable(deprecated)]
     #[configurable(metadata(
         docs::additional_props_description = "An HTTP request header and it's value."
     ))]
-    pub headers: Option<IndexMap<String, String>>,
+    pub(super) headers: Option<IndexMap<String, String>>,
 
     #[configurable(derived)]
     #[serde(default)]
-    pub compression: Compression,
+    pub(super) compression: Compression,
 
     #[serde(flatten)]
-    pub encoding: EncodingConfigWithFraming,
+    pub(super) encoding: EncodingConfigWithFraming,
 
     /// A string to prefix the payload with.
     ///
@@ -70,7 +70,7 @@ pub struct HttpSinkConfig {
     /// If specified, the `payload_suffix` must also be specified and together they must produce a valid JSON object.
     #[configurable(metadata(docs::examples = "{\"data\":"))]
     #[serde(default)]
-    pub payload_prefix: String,
+    pub(super) payload_prefix: String,
 
     /// A string to suffix the payload with.
     ///
@@ -79,26 +79,26 @@ pub struct HttpSinkConfig {
     /// If specified, the `payload_prefix` must also be specified and together they must produce a valid JSON object.
     #[configurable(metadata(docs::examples = "}"))]
     #[serde(default)]
-    pub payload_suffix: String,
+    pub(super) payload_suffix: String,
 
     #[configurable(derived)]
     #[serde(default)]
-    pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
+    pub(super) batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: RequestConfig,
+    pub(super) request: RequestConfig,
 
     #[configurable(derived)]
-    pub tls: Option<TlsConfig>,
+    pub(super) tls: Option<TlsConfig>,
 
     #[configurable(derived)]
     #[serde(
         default,
         deserialize_with = "crate::serde::bool_or_struct",
         skip_serializing_if = "crate::serde::skip_serializing_if_default"
     )]
-    pub acknowledgements: AcknowledgementsConfig,
+    pub(super) acknowledgements: AcknowledgementsConfig,
 }
 
 /// HTTP method.
@@ -110,7 +110,7 @@ pub struct HttpSinkConfig {
 #[derive(Clone, Copy, Debug, Derivative, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]
 #[derivative(Default)]
-pub enum HttpMethod {
+pub(super) enum HttpMethod {
     /// GET.
     Get,
 
@@ -277,14 +277,14 @@ impl SinkConfig for HttpSinkConfig {
                 .to_string()
         });
 
-        let http_service_request_builder = HttpSinkRequestBuilder {
-            uri: self.uri.with_default_parts(),
-            method: self.method,
-            auth: self.auth.choose_one(&self.uri.auth)?,
+        let http_service_request_builder = HttpSinkRequestBuilder::new(
+            self.uri.with_default_parts(),
+            self.method,
+            self.auth.choose_one(&self.uri.auth)?,
             headers,
             content_type,
             content_encoding,
-        };
+        );
 
         let service = HttpService::new(client, http_service_request_builder);
 

src/sinks/http/request_builder.rs

@@ -22,8 +22,6 @@ impl RequestBuilder<Vec<Event>> for HttpRequestBuilder {
 
     fn compression(&self) -> Compression {
         self.compression
-        // Compression is handled in the Service
-        //Compression::None
     }
 
     fn encoder(&self) -> &Self::Encoder {

src/sinks/http/service.rs

@@ -23,17 +23,16 @@ use crate::{
 
 use super::config::HttpMethod;
 
-/// Request type for use in `RequestBuilder` implementations of HTTP stream sinks.
 #[derive(Clone)]
-pub struct HttpRequest {
-    pub payload: Bytes,
-    pub finalizers: EventFinalizers,
-    pub request_metadata: RequestMetadata,
+pub(super) struct HttpRequest {
+    payload: Bytes,
+    finalizers: EventFinalizers,
+    request_metadata: RequestMetadata,
 }
 
 impl HttpRequest {
     /// Creates a new `HttpRequest`.
-    pub fn new(
+    pub(super) fn new(
         payload: Bytes,
         finalizers: EventFinalizers,
         request_metadata: RequestMetadata,
@@ -69,10 +68,10 @@ impl ByteSizeOf for HttpRequest {
 }
 
 /// Response type for use in the `Service` implementation of HTTP stream sinks.
-pub struct HttpResponse {
+pub(super) struct HttpResponse {
     pub(super) http_response: Response<Bytes>,
-    pub(super) events_byte_size: GroupedCountByteSize,
-    pub(super) raw_byte_size: usize,
+    events_byte_size: GroupedCountByteSize,
+    raw_byte_size: usize,
 }
 
 impl DriverResponse for HttpResponse {
@@ -95,19 +94,37 @@ impl DriverResponse for HttpResponse {
 
 #[derive(Debug, Clone)]
 pub(super) struct HttpSinkRequestBuilder {
-    pub(super) uri: UriSerde,
-    pub(super) method: HttpMethod,
-    pub(super) auth: Option<Auth>,
-    pub(super) headers: IndexMap<HeaderName, HeaderValue>,
-    pub(super) content_type: Option<String>,
-    pub(super) content_encoding: Option<String>,
+    uri: UriSerde,
+    method: HttpMethod,
+    auth: Option<Auth>,
+    headers: IndexMap<HeaderName, HeaderValue>,
+    content_type: Option<String>,
+    content_encoding: Option<String>,
 }
 
 impl HttpSinkRequestBuilder {
+    /// Creates a new `HttpSinkRequestBuilder`
+    pub(super) const fn new(
+        uri: UriSerde,
+        method: HttpMethod,
+        auth: Option<Auth>,
+        headers: IndexMap<HeaderName, HeaderValue>,
+        content_type: Option<String>,
+        content_encoding: Option<String>,
+    ) -> Self {
+        Self {
+            uri,
+            method,
+            auth,
+            headers,
+            content_type,
+            content_encoding,
+        }
+    }
+
     fn build(&self, body: Bytes) -> Request<Bytes> {
         let method: Method = self.method.into();
         let uri: Uri = self.uri.uri.clone();
-
         let mut builder = Request::builder().method(method).uri(uri);
 
         if let Some(content_type) = &self.content_type {
@@ -142,14 +159,12 @@ impl HttpSinkRequestBuilder {
 
 #[derive(Clone)]
 pub(super) struct HttpService {
-    // pub(crate) batch_service:
-    //     HttpBatchService<Ready<Result<http::Request<Bytes>, crate::Error>>, HttpRequest>,
     batch_service:
         HttpBatchService<BoxFuture<'static, Result<Request<Bytes>, crate::Error>>, HttpRequest>,
 }
 
 impl HttpService {
-    pub fn new(
+    pub(super) fn new(
         http_client: HttpClient<Body>,
         http_request_builder: HttpSinkRequestBuilder,
     ) -> Self {

src/sinks/http/sink.rs

@@ -33,8 +33,7 @@ where
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         let service = ServiceBuilder::new().service(self.service);
         input
-            // .batched(self.batch_settings.into_byte_size_config())
-            // Batch the input stream with size calculation dependent on the configured codec
+            // Batch the input stream with size calculation based on the configured codec
             .batched(self.batch_settings.into_item_size_config(HttpBatchSizer {
                 encoder: self.request_builder.encoder.encoder.clone(),
             }))