<commit_before>use client_died_error_holder::ClientConnDiedType;
use client_died_error_holder::ClientDiedErrorHolder;
use common::conn_command_channel::ConnCommandSender;
use common::increase_in_window::IncreaseInWindow;
use common::stream_from_network::StreamFromNetwork;
use common::stream_queue_sync::stream_queue_sync;
use server::increase_in_window::ServerIncreaseInWindow;
use server::stream_handler::ServerStreamHandler;
use server::stream_handler::ServerStreamHandlerHolder;
use server::types::ServerTypes;
use Headers;
use HttpStreamAfterHeaders;
use StreamId;

pub struct ServerRequest<'a> {
    /// Request headers
    pub headers: Headers,
    /// True if requests ends with headers
    pub end_stream: bool,
    pub(crate) stream_id: StreamId,
    /// Stream in window size at the moment of request start
    pub(crate) in_window_size: u32,
    pub(crate) stream_handler: &'a mut Option<ServerStreamHandlerHolder>,
    pub(crate) conn_died_error_holder: &'a ClientDiedErrorHolder<ClientConnDiedType>,
    pub(crate) to_write_tx: &'a ConnCommandSender<ServerTypes>,
}

impl<'a> ServerRequest<'a> {
    pub fn make_stream(&mut self) -> HttpStreamAfterHeaders {
        if self.end_stream {
            HttpStreamAfterHeaders::empty()
        } else {
            let client_died_error_holder = self.conn_died_error_holder.clone();
            let in_window_size = self.in_window_size;
            self.register_stream_handler(|increase_in_window| {
                let (inc_tx, inc_rx) = stream_queue_sync(client_died_error_holder);
                let stream_from_network = StreamFromNetwork {
                    rx: inc_rx,
                    increase_in_window: increase_in_window.0,
                    in_window_size,
                };

                (
                    inc_tx,
                    HttpStreamAfterHeaders::from_parts(stream_from_network),
                )
            })
        }
    }

    pub fn register_stream_handler<F, H, R>(&mut self, f: F) -> R
    where
        F: FnOnce(ServerIncreaseInWindow) -> (H, R),
        H: ServerStreamHandler,
    {
        assert!(self.stream_handler.is_none());
        let increase_window = ServerIncreaseInWindow(IncreaseInWindow {
            stream_id: self.stream_id,
            to_write_tx: self.to_write_tx.clone(),
        });
        let (h, r) = f(increase_window);
        *self.stream_handler = Some(ServerStreamHandlerHolder(Box::new(h)));<commit_msg>Supply current window size to register_stream_handler callback<commit_after>use client_died_error_holder::ClientConnDiedType;
use client_died_error_holder::ClientDiedErrorHolder;
use common::conn_command_channel::ConnCommandSender;
use common::increase_in_window::IncreaseInWindow;
use common::stream_from_network::StreamFromNetwork;
use common::stream_queue_sync::stream_queue_sync;
use server::increase_in_window::ServerIncreaseInWindow;
use server::stream_handler::ServerStreamHandler;
use server::stream_handler::ServerStreamHandlerHolder;
use server::types::ServerTypes;
use Headers;
use HttpStreamAfterHeaders;
use StreamId;

pub struct ServerRequest<'a> {
    /// Request headers
    pub headers: Headers,
    /// True if requests ends with headers
    pub end_stream: bool,
    pub(crate) stream_id: StreamId,
    /// Stream in window size at the moment of request start
    pub(crate) in_window_size: u32,
    pub(crate) stream_handler: &'a mut Option<ServerStreamHandlerHolder>,
    pub(crate) conn_died_error_holder: &'a ClientDiedErrorHolder<ClientConnDiedType>,
    pub(crate) to_write_tx: &'a ConnCommandSender<ServerTypes>,
}

impl<'a> ServerRequest<'a> {
    pub fn make_stream(&mut self) -> HttpStreamAfterHeaders {
        if self.end_stream {
            HttpStreamAfterHeaders::empty()
        } else {
            let client_died_error_holder = self.conn_died_error_holder.clone();
            self.register_stream_handler(|increase_in_window, in_window_size| {
                let (inc_tx, inc_rx) = stream_queue_sync(client_died_error_holder);
                let stream_from_network = StreamFromNetwork {
                    rx: inc_rx,
                    increase_in_window: increase_in_window.0,
                    in_window_size,
                };

                (
                    inc_tx,
                    HttpStreamAfterHeaders::from_parts(stream_from_network),
                )
            })
        }
    }

    pub fn register_stream_handler<F, H, R>(&mut self, f: F) -> R
    where
        F: FnOnce(ServerIncreaseInWindow, u32) -> (H, R),
        H: ServerStreamHandler,
    {
        assert!(self.stream_handler.is_none());
        let increase_window = ServerIncreaseInWindow(IncreaseInWindow {
            stream_id: self.stream_id,
            to_write_tx: self.to_write_tx.clone(),
        });
        let (h, r) = f(increase_window, self.in_window_size);
        *self.stream_handler = Some(ServerStreamHandlerHolder(Box::new(h)));
{'fraction_non_alphanumeric': 0.09389576150882174, 'fraction_numerical': 0.0016223889677550193, 'mean_word_length': 3.1585160202360876, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2530572', 'n_tokens_mistral': 1489, 'n_tokens_neox': 1447, 'n_words': 273}