<commit_before># -*- coding: utf-8 -*-
"""
hyper/http20/stream
~~~~~~~~~~~~~~~~~~~

Objects that make up the stream-level abstraction of hyper's HTTP/2.0 support.

These objects are not expected to be part of the public HTTP/2.0 API: they're
intended purely for use inside hyper's HTTP/2.0 abstraction.

Conceptually, a single HTTP/2.0 connection is made up of many streams: each
stream is an independent, bi-directional sequence of HTTP headers and data.
Each stream is identified by a monotonically increasing integer, assigned to
the stream by the endpoint that initiated the stream.
"""
from .frame import (
    FRAME_MAX_LEN, HeadersFrame, DataFrame, WindowUpdateFrame,
    ContinuationFrame,
)
from .response import HTTP20Response
import collections


# Define a set of states for a HTTP/2.0 stream.
STATE_IDLE               = 0
STATE_OPEN               = 1
STATE_HALF_CLOSED_LOCAL  = 2
STATE_HALF_CLOSED_REMOTE = 3
STATE_CLOSED             = 4


# Define the largest chunk of data we'll send in one go. Realistically, we
# should take the MSS into account but that's pretty dull, so let's just say
# 1kB and call it a day.
MAX_CHUNK = 1024


class Stream(object):
    """
    A single HTTP/2.0 stream.

    A stream is an independent, bi-directional sequence of HTTP headers and
    data. Each stream is identified by a single integer. From a HTTP
    perspective, a stream _approximately_ matches a single request-response
    pair.
    """
    def __init__(self,
                 stream_id,
                 data_cb,
                 recv_cb,
                 header_encoder,
                 header_decoder):
        self.stream_id = stream_id
        self.state = STATE_IDLE
        self.headers = []
        self._queued_frames = collections.deque()

        # There are two flow control windows: one for data we're sending,
        # one for data being sent to us.
        self._out_flow_control_window = 65535
        self._in_flow_control_window = 65535

        # This is the callback handed to the stream by its parent connection.
        # It is called when the stream wants to send data. It expects to
        # receive a list of frames that will be automatically serialized.
        self._data_cb = data_cb

        # Similarly, this is a callback that reads one frame off the
        # connection.
        self._recv_cb = recv_cb

        # A reference to the header encoder and decoder objects belonging to
        # the parent connection.
        self._encoder = header_encoder
        self._decoder = header_decoder

    def add_header(self, name, value):
        """
        Adds a single HTTP header to the headers to be sent on the request.
        """
        self.headers.append((name, value))

    def send_data(self, data, final):
        """
        Send some data on the stream. If this is the end of the data to be
        sent, the ``final`` flag _must_ be set to True. If no data is to be
        sent, set ``data`` to ``None``.
        """
        # Define a utility iterator for file objects.
        def file_iterator(fobj):
            while True:
                data = fobj.read(MAX_CHUNK)
                yield data
                if len(data) < MAX_CHUNK:
                    break

        # Build the appropriate iterator for the data, in chunks of CHUNK_SIZE.
        if hasattr(data, 'read'):
            chunks = file_iterator(data)
        else:
            chunks = (data[i:i+MAX_CHUNK]
                      for i in range(0, len(data), MAX_CHUNK))

        for chunk in chunks:
            self._send_chunk(chunk, final)

    def _read(self, amt=None):
        """
        Read data from the stream. Unlike a normal read behaviour, this
        function returns _at least_ ``amt`` data, but may return more.
        """
        assert self.state in (STATE_OPEN, STATE_HALF_CLOSED_LOCAL)

        def listlen(list):
            return sum(map(len, list))

        data = []

        # Begin by processing frames off the queue.
        while amt is None or listlen(data) < amt:
            try:
                frame = self._queued_frames.popleft()
            except IndexError:
                # No frames on the queue. Try to read one and try again.
                self._recv_cb()
                continue

            # All queued frames at this point should be data frames.
            assert isinstance(frame, DataFrame)

            # Append the data to the buffer.
            data.append(frame.data)
<commit_msg>Allow read() calls on exhausted streams.<commit_after># -*- coding: utf-8 -*-
"""
hyper/http20/stream
~~~~~~~~~~~~~~~~~~~

Objects that make up the stream-level abstraction of hyper's HTTP/2.0 support.

These objects are not expected to be part of the public HTTP/2.0 API: they're
intended purely for use inside hyper's HTTP/2.0 abstraction.

Conceptually, a single HTTP/2.0 connection is made up of many streams: each
stream is an independent, bi-directional sequence of HTTP headers and data.
Each stream is identified by a monotonically increasing integer, assigned to
the stream by the endpoint that initiated the stream.
"""
from .frame import (
    FRAME_MAX_LEN, HeadersFrame, DataFrame, WindowUpdateFrame,
    ContinuationFrame,
)
from .response import HTTP20Response
import collections


# Define a set of states for a HTTP/2.0 stream.
STATE_IDLE               = 0
STATE_OPEN               = 1
STATE_HALF_CLOSED_LOCAL  = 2
STATE_HALF_CLOSED_REMOTE = 3
STATE_CLOSED             = 4


# Define the largest chunk of data we'll send in one go. Realistically, we
# should take the MSS into account but that's pretty dull, so let's just say
# 1kB and call it a day.
MAX_CHUNK = 1024


class Stream(object):
    """
    A single HTTP/2.0 stream.

    A stream is an independent, bi-directional sequence of HTTP headers and
    data. Each stream is identified by a single integer. From a HTTP
    perspective, a stream _approximately_ matches a single request-response
    pair.
    """
    def __init__(self,
                 stream_id,
                 data_cb,
                 recv_cb,
                 header_encoder,
                 header_decoder):
        self.stream_id = stream_id
        self.state = STATE_IDLE
        self.headers = []
        self._queued_frames = collections.deque()

        # There are two flow control windows: one for data we're sending,
        # one for data being sent to us.
        self._out_flow_control_window = 65535
        self._in_flow_control_window = 65535

        # This is the callback handed to the stream by its parent connection.
        # It is called when the stream wants to send data. It expects to
        # receive a list of frames that will be automatically serialized.
        self._data_cb = data_cb

        # Similarly, this is a callback that reads one frame off the
        # connection.
        self._recv_cb = recv_cb

        # A reference to the header encoder and decoder objects belonging to
        # the parent connection.
        self._encoder = header_encoder
        self._decoder = header_decoder

    def add_header(self, name, value):
        """
        Adds a single HTTP header to the headers to be sent on the request.
        """
        self.headers.append((name, value))

    def send_data(self, data, final):
        """
        Send some data on the stream. If this is the end of the data to be
        sent, the ``final`` flag _must_ be set to True. If no data is to be
        sent, set ``data`` to ``None``.
        """
        # Define a utility iterator for file objects.
        def file_iterator(fobj):
            while True:
                data = fobj.read(MAX_CHUNK)
                yield data
                if len(data) < MAX_CHUNK:
                    break

        # Build the appropriate iterator for the data, in chunks of CHUNK_SIZE.
        if hasattr(data, 'read'):
            chunks = file_iterator(data)
        else:
            chunks = (data[i:i+MAX_CHUNK]
                      for i in range(0, len(data), MAX_CHUNK))

        for chunk in chunks:
            self._send_chunk(chunk, final)

    def _read(self, amt=None):
        """
        Read data from the stream. Unlike a normal read behaviour, this
        function returns _at least_ ``amt`` data, but may return more.
        """
        if self.state == STATE_CLOSED:
            return b''

        assert self.state in (STATE_OPEN, STATE_HALF_CLOSED_LOCAL)

        def listlen(list):
            return sum(map(len, list))

        data = []

        # Begin by processing frames off the queue.
        while amt is None or listlen(data) < amt:
            try:
                frame = self._queued_frames.popleft()
            except IndexError:
                # No frames on the queue. Try to read one and try again.
                self._recv_cb()
                continue

            # All queued frames at this point should be data frames.
            assert isinstance(frame, DataFrame)

            # Append the data to the buffer.
            data.append(frame.data)

{'fraction_non_alphanumeric': 0.07114405842000443, 'fraction_numerical': 0.008408940030980305, 'mean_word_length': 2.0671869697997964, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, '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': '5411236', 'n_tokens_mistral': 2510, 'n_tokens_neox': 2384, 'n_words': 988}