# Path: streamcorpus_pipeline/_exceptions.py
# class ConfigurationError(PipelineBaseException, yakonfig.ConfigurationError):
#     pass
# 
# Path: streamcorpus_pipeline/_spinn3r/protoStream_pb2.py
# class ProtoStreamDelimiter(message.Message):
#   __metaclass__ = reflection.GeneratedProtocolMessageType
#   DESCRIPTOR = _PROTOSTREAMDELIMITER
#   
#   # @@protoc_insertion_point(class_scope:protoStream.ProtoStreamDelimiter)
# 
# class ProtoStreamHeader(message.Message):
#   __metaclass__ = reflection.GeneratedProtocolMessageType
#   DESCRIPTOR = _PROTOSTREAMHEADER
#   
#   # @@protoc_insertion_point(class_scope:protoStream.ProtoStreamHeader)
# 
# Path: streamcorpus_pipeline/stages.py
# class Configured(object):
#     '''Any object containing a configuration.
# 
#     The configuration is passed to the constructor.
# 
#     .. attribute:: config
# 
#         The configuration dictionary for this object.
# 
#     '''
#     def __init__(self, config, *args, **kwargs):
#         super(Configured, self).__init__(*args, **kwargs)
#         self.config = config


import itertools
import json
import logging
import zlib
import streamcorpus
from StringIO import StringIO
from google.protobuf.internal.decoder import _DecodeVarint
from streamcorpus_pipeline._exceptions import ConfigurationError
from streamcorpus_pipeline._spinn3r.protoStream_pb2 \
    import ProtoStreamDelimiter, ProtoStreamHeader
from streamcorpus_pipeline._spinn3r.spinn3rApi_pb2 import Entry
from streamcorpus_pipeline.stages import Configured

        ``f``
          file-like object to read
        ``entry_type``
          type of objects in this feed

        """
        super(ProtoStreamReader, self).__init__()
        self.f = f
        self._prefix = ''
        self._header = None
        self._entry_type = entry_type

    def _unread(self, b):
        """Push byte array 'b' on to the front of the prefix list; this
        undoes part of a _read().  This bytes will be the next to read."""
        self._prefix = b + self._prefix

    def _read(self, n):
        """Read (up to) 'n' bytes from the underlying file.  If any bytes
        have been pushed in with _unread() those are returned first."""
        if n <= len(self._prefix):
            # the read can be fulfilled entirely from the prefix
            result = self._prefix[:n]
            self._prefix = self._prefix[n:]
            return result
        # otherwise we need to read some
        n -= len(self._prefix)
        result = self._prefix + self.f.read(n)
        self._prefix = ""

# Path: streamcorpus_pipeline/_exceptions.py
# class ConfigurationError(PipelineBaseException, yakonfig.ConfigurationError):
#     pass
# 
# Path: streamcorpus_pipeline/_spinn3r/protoStream_pb2.py
# class ProtoStreamDelimiter(message.Message):
#   __metaclass__ = reflection.GeneratedProtocolMessageType
#   DESCRIPTOR = _PROTOSTREAMDELIMITER
#   
#   # @@protoc_insertion_point(class_scope:protoStream.ProtoStreamDelimiter)
# 
# class ProtoStreamHeader(message.Message):
#   __metaclass__ = reflection.GeneratedProtocolMessageType
#   DESCRIPTOR = _PROTOSTREAMHEADER
#   
#   # @@protoc_insertion_point(class_scope:protoStream.ProtoStreamHeader)
# 
# Path: streamcorpus_pipeline/stages.py
# class Configured(object):
#     '''Any object containing a configuration.
# 
#     The configuration is passed to the constructor.
# 
#     .. attribute:: config
# 
#         The configuration dictionary for this object.
# 
#     '''
#     def __init__(self, config, *args, **kwargs):
#         super(Configured, self).__init__(*args, **kwargs)
#         self.config = config


# Path: streamcorpus_pipeline/_spinn3r_feed_storage.py
import itertools
import json
import logging
import zlib
import streamcorpus
from StringIO import StringIO
from google.protobuf.internal.decoder import _DecodeVarint
from streamcorpus_pipeline._exceptions import ConfigurationError
from streamcorpus_pipeline._spinn3r.protoStream_pb2 \
    import ProtoStreamDelimiter, ProtoStreamHeader
from streamcorpus_pipeline._spinn3r.spinn3rApi_pb2 import Entry
from streamcorpus_pipeline.stages import Configured


        ``f``
          file-like object to read
        ``entry_type``
          type of objects in this feed

        """
        super(ProtoStreamReader, self).__init__()
        self.f = f
        self._prefix = ''
        self._header = None
        self._entry_type = entry_type

    def _unread(self, b):
        """Push byte array 'b' on to the front of the prefix list; this
        undoes part of a _read().  This bytes will be the next to read."""
        self._prefix = b + self._prefix

    def _read(self, n):
        """Read (up to) 'n' bytes from the underlying file.  If any bytes
        have been pushed in with _unread() those are returned first."""
        if n <= len(self._prefix):
            # the read can be fulfilled entirely from the prefix
            result = self._prefix[:n]
            self._prefix = self._prefix[n:]
            return result
        # otherwise we need to read some
        n -= len(self._prefix)
        result = self._prefix + self.f.read(n)
        self._prefix = ""

        return result