<s> """<STR_LIT>""" <EOL> __all__ = [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> import logging <EOL> import struct <EOL> from mapreduce . third_party import crc32c <EOL> from mapreduce import errors <EOL> _BLOCK_SIZE = <NUM_LIT:32> * <NUM_LIT> <EOL> _HEADER_FORMAT = '<STR_LIT>' <EOL> _HEADER_LENGTH = struct . calcsize ( _HEADER_FORMAT ) <EOL> _RECORD_TYPE_NONE = <NUM_LIT:0> <EOL> _RECORD_TYPE_FULL = <NUM_LIT:1> <EOL> _RECORD_TYPE_FIRST = <NUM_LIT:2> <EOL> _RECORD_TYPE_MIDDLE = <NUM_LIT:3> <EOL> _RECORD_TYPE_LAST = <NUM_LIT:4> <EOL> _CRC_MASK_DELTA = <NUM_LIT> <EOL> def _mask_crc ( crc ) : <EOL> """<STR_LIT>""" <EOL> return ( ( ( crc >> <NUM_LIT:15> ) | ( crc << <NUM_LIT> ) ) + _CRC_MASK_DELTA ) & <NUM_LIT> L <EOL> def _unmask_crc ( masked_crc ) : <EOL> """<STR_LIT>""" <EOL> rot = ( masked_crc - _CRC_MASK_DELTA ) & <NUM_LIT> L <EOL> return ( ( rot >> <NUM_LIT> ) | ( rot << <NUM_LIT:15> ) ) & <NUM_LIT> L <EOL> class RecordsWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , writer ) : <EOL> """<STR_LIT>""" <EOL> self . __writer = writer <EOL> self . __position = <NUM_LIT:0> <EOL> def __write_record ( self , record_type , data ) : <EOL> """<STR_LIT>""" <EOL> length = len ( data ) <EOL> crc = crc32c . crc_update ( crc32c . CRC_INIT , [ record_type ] ) <EOL> crc = crc32c . crc_update ( crc , data ) <EOL> crc = crc32c . crc_finalize ( crc ) <EOL> self . __writer . write ( <EOL> struct . pack ( _HEADER_FORMAT , _mask_crc ( crc ) , length , record_type ) ) <EOL> self . __writer . write ( data ) <EOL> self . __position += _HEADER_LENGTH + length <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> block_remaining = _BLOCK_SIZE - self . __position % _BLOCK_SIZE <EOL> if block_remaining < _HEADER_LENGTH : <EOL> self . __writer . write ( '<STR_LIT:\x00>' * block_remaining ) <EOL> self . __position += block_remaining <EOL> block_remaining = _BLOCK_SIZE <EOL> if block_remaining < len ( data ) + _HEADER_LENGTH : <EOL> first_chunk = data [ : block_remaining - _HEADER_LENGTH ] <EOL> self . __write_record ( _RECORD_TYPE_FIRST , first_chunk ) <EOL> data = data [ len ( first_chunk ) : ] <EOL> while True : <EOL> block_remaining = _BLOCK_SIZE - self . __position % _BLOCK_SIZE <EOL> if block_remaining >= len ( data ) + _HEADER_LENGTH : <EOL> self . __write_record ( _RECORD_TYPE_LAST , data ) <EOL> break <EOL> else : <EOL> chunk = data [ : block_remaining - _HEADER_LENGTH ] <EOL> self . __write_record ( _RECORD_TYPE_MIDDLE , chunk ) <EOL> data = data [ len ( chunk ) : ] <EOL> else : <EOL> self . __write_record ( _RECORD_TYPE_FULL , data ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , atype , value , traceback ) : <EOL> self . close ( ) <EOL> def close ( self ) : <EOL> pass <EOL> def _pad_block ( self ) : <EOL> """<STR_LIT>""" <EOL> pad_length = _BLOCK_SIZE - self . __position % _BLOCK_SIZE <EOL> if pad_length and pad_length != _BLOCK_SIZE : <EOL> self . __writer . write ( '<STR_LIT:\x00>' * pad_length ) <EOL> self . __position += pad_length <EOL> class RecordsReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reader ) : <EOL> """<STR_LIT>""" <EOL> self . __reader = reader <EOL> def __try_read_record ( self ) : <EOL> """<STR_LIT>""" <EOL> block_remaining = _BLOCK_SIZE - self . __reader . tell ( ) % _BLOCK_SIZE <EOL> if block_remaining < _HEADER_LENGTH : <EOL> return ( '<STR_LIT>' , _RECORD_TYPE_NONE ) <EOL> header = self . __reader . read ( _HEADER_LENGTH ) <EOL> if len ( header ) != _HEADER_LENGTH : <EOL> raise EOFError ( '<STR_LIT>' % <EOL> ( len ( header ) , _HEADER_LENGTH ) ) <EOL> ( masked_crc , length , record_type ) = struct . unpack ( _HEADER_FORMAT , header ) <EOL> crc = _unmask_crc ( masked_crc ) <EOL> if length + _HEADER_LENGTH > block_remaining : <EOL> raise errors . InvalidRecordError ( '<STR_LIT>' ) <EOL> data = self . __reader . read ( length ) <EOL> if len ( data ) != length : <EOL> raise EOFError ( '<STR_LIT>' % <EOL> ( length , len ( data ) ) ) <EOL> if record_type == _RECORD_TYPE_NONE : <EOL> return ( '<STR_LIT>' , record_type ) <EOL> actual_crc = crc32c . crc_update ( crc32c . CRC_INIT , [ record_type ] ) <EOL> actual_crc = crc32c . crc_update ( actual_crc , data ) <EOL> actual_crc = crc32c . crc_finalize ( actual_crc ) <EOL> if actual_crc != crc : <EOL> raise errors . InvalidRecordError ( '<STR_LIT>' ) <EOL> return ( data , record_type ) <EOL> def __sync ( self ) : <EOL> """<STR_LIT>""" <EOL> pad_length = _BLOCK_SIZE - self . __reader . tell ( ) % _BLOCK_SIZE <EOL> if pad_length and pad_length != _BLOCK_SIZE : <EOL> data = self . __reader . read ( pad_length ) <EOL> if len ( data ) != pad_length : <EOL> raise EOFError ( '<STR_LIT>' % <EOL> ( len ( data ) , pad_length ) ) <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> data = None <EOL> while True : <EOL> last_offset = self . tell ( ) <EOL> try : <EOL> ( chunk , record_type ) = self . __try_read_record ( ) <EOL> if record_type == _RECORD_TYPE_NONE : <EOL> self . __sync ( ) <EOL> elif record_type == _RECORD_TYPE_FULL : <EOL> if data is not None : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , last_offset ) <EOL> return chunk <EOL> elif record_type == _RECORD_TYPE_FIRST : <EOL> if data is not None : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , last_offset ) <EOL> data = chunk <EOL> elif record_type == _RECORD_TYPE_MIDDLE : <EOL> if data is None : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , last_offset ) <EOL> else : <EOL> data += chunk <EOL> elif record_type == _RECORD_TYPE_LAST : <EOL> if data is None : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , last_offset ) <EOL> else : <EOL> result = data + chunk <EOL> data = None <EOL> return result <EOL> else : <EOL> raise errors . InvalidRecordError ( <EOL> "<STR_LIT>" % record_type ) <EOL> except errors . InvalidRecordError , e : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , last_offset , e ) <EOL> data = None <EOL> self . __sync ( ) <EOL> def __iter__ ( self ) : <EOL> try : <EOL> while True : <EOL> yield self . read ( ) <EOL> except EOFError : <EOL> pass <EOL> def tell ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __reader . tell ( ) <EOL> def seek ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . __reader . seek ( * args , ** kwargs ) </s>
