<commit_before>"""Messagepack implementation of pystorm serializer"""

from __future__ import absolute_import, print_function, unicode_literals

import io
import os

import msgpack

from ..exceptions import StormWentAwayError
from .serializer import Serializer


class MsgpackSerializer(Serializer):

    def __init__(self, input_stream, output_stream, reader_lock, writer_lock):
        super(MsgpackSerializer, self).__init__(input_stream,
                                                self._raw_stream(output_stream),
                                                reader_lock, writer_lock)
        self._messages = self._messages_generator()

    @staticmethod
    def _raw_stream(stream):
        """Returns the raw buffer used by stream, so we can write bytes."""
        if hasattr(stream, 'buffer'):
            return stream.buffer
        else:
            return stream

    def _messages_generator(self):
        unpacker = msgpack.Unpacker()
        while True:
            # f.read(n) on sys.stdin blocks until n bytes are read, causing
            # serializer to hang.
            # os.read(fileno, n) will block if there is nothing to read, but will
            # return as soon as it is able to read at most n bytes.
            try:
                line = os.read(self.input_stream.fileno(), 1024 ** 2)
            except io.UnsupportedOperation:
                line = self.input_stream.read(1024 ** 2)
            if not line:
                # Handle EOF, which usually means Storm went away
                raise StormWentAwayError()<commit_msg>Add missing _reader_lock in msgpack_serializer<commit_after>"""Messagepack implementation of pystorm serializer"""

from __future__ import absolute_import, print_function, unicode_literals

import io
import os

import msgpack

from ..exceptions import StormWentAwayError
from .serializer import Serializer


class MsgpackSerializer(Serializer):

    CHUNK_SIZE = 1024 ** 2

    def __init__(self, input_stream, output_stream, reader_lock, writer_lock):
        super(MsgpackSerializer, self).__init__(input_stream,
                                                self._raw_stream(output_stream),
                                                reader_lock, writer_lock)
        self._messages = self._messages_generator()

    @staticmethod
    def _raw_stream(stream):
        """Returns the raw buffer used by stream, so we can write bytes."""
        if hasattr(stream, 'buffer'):
            return stream.buffer
        else:
            return stream

    def _messages_generator(self):
        unpacker = msgpack.Unpacker()
        while True:
            # f.read(n) on sys.stdin blocks until n bytes are read, causing
            # serializer to hang.
            # os.read(fileno, n) will block if there is nothing to read, but will
            # return as soon as it is able to read at most n bytes.
            with self._reader_lock:
                try:
                    line = os.read(self.input_stream.fileno(), self.CHUNK_SIZE)
                except io.UnsupportedOperation:
                    line = self.input_stream.read(self.CHUNK_SIZE)
            if not line:
                # Handle EOF, which usually means Storm went away
                raise StormWentAwayError()
{'fraction_non_alphanumeric': 0.06887052341597796, 'fraction_numerical': 0.004591368227731864, 'mean_word_length': 2.0714285714285716, 'pattern_counts': {'":': 0, '<': 3, '<?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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5989218', 'n_tokens_mistral': 848, 'n_tokens_neox': 815, 'n_words': 266}