<commit_before># -*- coding: utf-8 -*-

import helpers
import looping
import collections
# import flv

class StreamSource(looping.BaseIOEventHandler):

    def __init__(self, server, sock, address, request_parser):
        self.server = server
        self.sock = sock
        self.address = address
        self.request_parser = request_parser
        self.path = self.request_parser.request_path

    def publish_packet(self, packet):
        self.server.publish_packet(self, packet)

    def new_client(self, client):
        # Do nothing by default
        pass

class BufferedRawSource(StreamSource):

    # Approximately one packet/s for a 64 kb/s stream
    PACKET_SIZE = 8192

    def __init__(self, server, sock, address, request_parser):
        StreamSource.__init__(self, server, sock, address, request_parser)
        self.buffer_data = request_parser.body

    def publish_packet(self, packet):
        self.buffer_data = self.buffer_data + packet
        if len(self.buffer_data) >= self.PACKET_SIZE:
            StreamSource.publish_packet(self, self.buffer_data)
            self.buffer_data = ''

    def handle_event(self, eventmask):
        if eventmask & looping.POLLIN:
            while True:
                packet = helpers.handle_eagain(self.sock.recv, self.PACKET_SIZE)
                if packet == None:
                    # EAGAIN
                    break
                elif packet == b'':
                    # End of stream
                    print 'End of stream for %s, %s' % (self.sock, self.address)
                    self.server.remove_source(self)
                    # FIXME: publish "EOS" packet
                    break
                else:
                    self.publish_packet(packet)
        else:
            print 'Unexpected eventmask %s' % (eventmask)

class FLVSource(StreamSource):

    def __init__(self, sock, server, address, request_parser):
        StreamSource.__init__(self, sock, server, address, request_parser)
        self.stream_header = None
        self.initial_tags = []
        self.current_output_buffer = None
<commit_msg>Added support for an initial data burst on client connection<commit_after># -*- coding: utf-8 -*-

import helpers
import looping
import collections
import math
# import flv

class StreamSource(looping.BaseIOEventHandler):

    def __init__(self, server, sock, address, request_parser):
        self.server = server
        self.sock = sock
        self.address = address
        self.request_parser = request_parser
        self.path = self.request_parser.request_path

    def publish_packet(self, packet):
        self.server.publish_packet(self, packet)

    def new_client(self, client):
        # Do nothing by default
        pass

class BufferedRawSource(StreamSource):

    # Incoming maximum buffer size
    RECV_BUFFER_SIZE = 64 * 2**10

    # Temporary buffer size
    TEMP_BUFFER_SIZE = 4 * 2**10

    # Size of initial data burst for clients
    BURST_SIZE = 32 * 2**10

    def __init__(self, server, sock, address, request_parser):
        StreamSource.__init__(self, server, sock, address, request_parser)
        self.buffer_data = request_parser.body
        self.burst_packets = collections.deque([self.buffer_data], math.ceil(self.BURST_SIZE / self.TEMP_BUFFER_SIZE))

    def publish_packet(self, packet):
        self.buffer_data = self.buffer_data + packet
        if len(self.buffer_data) >= self.TEMP_BUFFER_SIZE:
            StreamSource.publish_packet(self, self.buffer_data)
            self.burst_packets.append(self.buffer_data)
            self.buffer_data = ''

    def new_client(self, client):
        for packet in self.burst_packets:
            client.add_packet(packet)

    def handle_event(self, eventmask):
        if eventmask & looping.POLLIN:
            while True:
                packet = helpers.handle_eagain(self.sock.recv, self.RECV_BUFFER_SIZE)
                if packet == None:
                    # EAGAIN
                    break
                elif packet == b'':
                    # End of stream
                    print 'End of stream for %s, %s' % (self.sock, self.address)
                    self.server.remove_source(self)
                    # FIXME: publish "EOS" packet
                    break
                else:
                    self.publish_packet(packet)
        else:
            print 'Unexpected eventmask %s' % (eventmask)

class FLVSource(StreamSource):

    def __init__(self, sock, server, address, request_parser):
        StreamSource.__init__(self, sock, server, address, request_parser)
        self.stream_header = None
        self.initial_tags = []
        self.current_output_buffer = None

{'fraction_non_alphanumeric': 0.08487950522499467, 'fraction_numerical': 0.004691831947110258, 'mean_word_length': 2.1818181818181817, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '851914', 'n_tokens_mistral': 1328, 'n_tokens_neox': 1267, 'n_words': 321}