You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: sink.rs
    path of file: ./repos/vector/src/sinks/http
    
    the code of the file until where you have to start completion: //! Implementation of the `http` sink.

use crate::sinks::{prelude::*, util::http::HttpRequest};

use super::{batch::HttpBatchSizer, request_builder::HttpRequestBuilder};

pub(super) struct HttpSink<S> {
    service: S,
    batch_settings: BatcherSettings,
    request_builder: HttpRequestBuilder,
}

impl<S> HttpSink<S>
where
    S: Service<HttpRequest<()>> + Send + 'static,
    S::Future: Send + 'static,
    S::Response: DriverResponse + Send + 'static,
    S::Error: std::fmt::Debug + Into<crate::Error> + Send,
{
    /// Creates a new `HttpSink`.
    pub(super) const fn new(
        service: S,
        batch_settings: BatcherSettings,
        request_builder: HttpRequestBuilder,
    ) -> Self {
        Self {
            service,
            batch_settings,
            request_builder,
        }
    }

    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
        input
            // Batch the input stream with size calculation based on the configured codec
            .bat
    
    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
        input
            // Batch the input stream with size calculation based on the configured codec
            .batched(self.batch_settings.as_item_size_config(HttpBatchSizer {
                encoder: self.request_builder.encoder.encoder.clone(),
            }))
            // Build requests with default concurrency limit.
            .request_builder(
                default_request_builder_concurrency_limit(),
                self.request_builder,
            )
            // Filter out any errors that occurred in the request building.
            .filter_map(|request| async move {
                match request {
                    Err(error) => {
                        emit!(SinkRequestBuildError { error });
                        None
                    }
                    Ok(req) => Some(req),
                }
            })
            // Generate the driver that will send requests and handle retries,
            // event finalization, and logging/internal metric reporting.
            .into_driver(self.service)
            .run()
            .await
    }