<s> """<STR_LIT>""" <EOL> from contextlib import contextmanager <EOL> from shutil import rmtree <EOL> from six . moves . queue import Queue <EOL> from tempfile import mkstemp <EOL> from threading import Thread <EOL> import logging as lg <EOL> import os <EOL> import os . path as osp <EOL> _logger = lg . getLogger ( __name__ ) <EOL> class HdfsError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message , * args ) : <EOL> super ( HdfsError , self ) . __init__ ( message % args if args else message ) <EOL> class AsyncWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , consumer ) : <EOL> self . _consumer = consumer <EOL> self . _queue = None <EOL> self . _reader = None <EOL> self . _err = None <EOL> _logger . debug ( '<STR_LIT>' , self ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . _consumer ) <EOL> def __enter__ ( self ) : <EOL> if self . _queue : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . _queue = Queue ( ) <EOL> self . _err = None <EOL> def consumer ( data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> self . _consumer ( data ) <EOL> except Exception as err : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> self . _err = err <EOL> finally : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> def reader ( queue ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> chunk = queue . get ( ) <EOL> if chunk is None : <EOL> break <EOL> yield chunk <EOL> self . _reader = Thread ( target = consumer , args = ( reader ( self . _queue ) , ) ) <EOL> self . _reader . start ( ) <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , traceback ) : <EOL> if exc_value : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> if self . _reader and self . _reader . is_alive ( ) : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> self . _queue . put ( None ) <EOL> self . _reader . join ( ) <EOL> if self . _err : <EOL> raise self . _err <EOL> else : <EOL> _logger . debug ( '<STR_LIT>' ) <EOL> self . _queue = None <EOL> def flush ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def write ( self , chunk ) : <EOL> """<STR_LIT>""" <EOL> self . _queue . put ( chunk ) <EOL> @ contextmanager <EOL> def temppath ( dpath = None ) : <EOL> """<STR_LIT>""" <EOL> ( desc , path ) = mkstemp ( dir = dpath ) <EOL> os . close ( desc ) <EOL> os . remove ( path ) <EOL> try : <EOL> _logger . debug ( '<STR_LIT>' , path ) <EOL> yield path <EOL> finally : <EOL> if osp . exists ( path ) : <EOL> if osp . isdir ( path ) : <EOL> rmtree ( path ) <EOL> _logger . debug ( '<STR_LIT>' , path ) <EOL> else : <EOL> os . remove ( path ) <EOL> _logger . debug ( '<STR_LIT>' , path ) <EOL> else : <EOL> _logger . debug ( '<STR_LIT>' , path ) </s>
