<s> from collections import deque , defaultdict , namedtuple <EOL> import storm <EOL> python_id = id <EOL> STORM_TUPLE = <NUM_LIT:0> <EOL> LIST = <NUM_LIT:1> <EOL> TUPLE = <NUM_LIT:2> <EOL> NAMEDTUPLE = <NUM_LIT:3> <EOL> class MockSpout ( storm . Spout ) : <EOL> def __init__ ( self , output_fields , data ) : <EOL> self . output_fields = output_fields <EOL> self . data = data <EOL> self . index = <NUM_LIT:0> <EOL> def declareOutputFields ( self ) : <EOL> return self . output_fields <EOL> def nextTuple ( self ) : <EOL> if self . index < len ( self . data ) : <EOL> storm . emit ( self . data [ self . index ] ) <EOL> self . index += <NUM_LIT:1> <EOL> class Mock ( object ) : <EOL> def __init__ ( self ) : <EOL> self . output_type = { } <EOL> self . pending = defaultdict ( deque ) <EOL> self . processed = defaultdict ( deque ) <EOL> self . emitter = None <EOL> def __enter__ ( self ) : <EOL> self . old_emit = storm . emit <EOL> storm . emit = self . emit <EOL> self . old_emitMany = storm . emitMany <EOL> storm . emitMany = self . emitMany <EOL> return self <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> storm . emit = self . old_emit <EOL> storm . emitMany = self . old_emitMany <EOL> def activate ( self , emitter ) : <EOL> self . emitter = emitter <EOL> if isinstance ( emitter , storm . Spout ) : <EOL> storm . MODE = storm . Spout <EOL> elif isinstance ( emitter , ( storm . Bolt , storm . BasicBolt ) ) : <EOL> storm . MODE = storm . Bolt <EOL> else : <EOL> assert False , "<STR_LIT>" <EOL> def emit ( self , * args , ** kwargs ) : <EOL> self . __emit ( * args , ** kwargs ) <EOL> def __emit ( self , * args , ** kwargs ) : <EOL> if storm . MODE == storm . Bolt : <EOL> self . emitBolt ( * args , ** kwargs ) <EOL> elif storm . MODE == storm . Spout : <EOL> self . emitSpout ( * args , ** kwargs ) <EOL> def emitMany ( self , * args , ** kwargs ) : <EOL> if storm . MODE == storm . Bolt : <EOL> self . emitManyBolt ( * args , ** kwargs ) <EOL> elif storm . MODE == storm . Spout : <EOL> self . emitManySpout ( * args , ** kwargs ) <EOL> def emitManyBolt ( self , tuples , stream = None , anchors = [ ] , directTask = None ) : <EOL> for t in tuples : <EOL> self . emitBolt ( t , stream , anchors , directTask ) <EOL> def emitManySpout ( self , tuples , stream = None , anchors = [ ] , directTask = None ) : <EOL> for t in tuples : <EOL> self . emitSpout ( t , stream , id , directTask ) <EOL> def emitter_id ( self , emitter = None ) : <EOL> if emitter is None : <EOL> emitter = self . emitter <EOL> return type ( emitter ) . __name__ , python_id ( emitter ) <EOL> def emitBolt ( self , tup , stream = None , anchors = [ ] , directTask = None ) : <EOL> self . pending [ self . emitter_id ( ) ] . append ( storm . Tuple ( id = None , component = None , stream = stream , task = directTask , values = tup ) ) <EOL> def emitSpout ( self , tup , stream = None , id = None , directTask = None ) : <EOL> self . pending [ self . emitter_id ( ) ] . append ( storm . Tuple ( id = id , component = None , stream = stream , task = directTask , values = tup ) ) <EOL> def read ( self , source_emitter ) : <EOL> emitter_id = self . emitter_id ( source_emitter ) <EOL> result = self . pending [ emitter_id ] . popleft ( ) <EOL> self . processed [ emitter_id ] . append ( result ) <EOL> return result <EOL> def get_output_type ( self , emitter ) : <EOL> emitter_id = self . emitter_id ( emitter ) <EOL> if emitter_id not in self . output_type : <EOL> self . output_type [ emitter_id ] = namedtuple ( '<STR_LIT>' % type ( emitter ) . __name__ , emitter . declareOutputFields ( ) ) <EOL> return self . output_type [ emitter_id ] <EOL> @ classmethod <EOL> def run_simple_topology ( cls , config , emitters , result_type = NAMEDTUPLE , max_spout_emits = None ) : <EOL> """<STR_LIT>""" <EOL> if config is not None : <EOL> for emitter in emitters : <EOL> emitter . initialize ( config , { } ) <EOL> with cls ( ) as self : <EOL> spout = emitters [ <NUM_LIT:0> ] <EOL> spout_id = self . emitter_id ( spout ) <EOL> old_length = - <NUM_LIT:1> <EOL> length = len ( self . pending [ spout_id ] ) <EOL> while length > old_length and ( max_spout_emits is None or length < max_spout_emits ) : <EOL> old_length = length <EOL> self . activate ( spout ) <EOL> spout . nextTuple ( ) <EOL> length = len ( self . pending [ spout_id ] ) <EOL> for i , bolt in enumerate ( emitters [ <NUM_LIT:1> : ] ) : <EOL> previous = emitters [ i ] <EOL> self . activate ( bolt ) <EOL> while len ( self . pending [ self . emitter_id ( previous ) ] ) > <NUM_LIT:0> : <EOL> bolt . process ( self . read ( previous ) ) <EOL> def make_storm_tuple ( t , emitter ) : <EOL> return t <EOL> def make_python_list ( t , emitter ) : <EOL> return list ( t . values ) <EOL> def make_python_tuple ( t , emitter ) : <EOL> return tuple ( t . values ) <EOL> def make_named_tuple ( t , emitter ) : <EOL> return self . get_output_type ( emitter ) ( * t . values ) <EOL> if result_type == STORM_TUPLE : <EOL> make = make_storm_tuple <EOL> elif result_type == LIST : <EOL> make = make_python_list <EOL> elif result_type == NAMEDTUPLE : <EOL> make = make_named_tuple <EOL> else : <EOL> assert False , '<STR_LIT>' % result_type <EOL> result_values = [ [ make ( t , emitter ) for t in self . processed [ self . emitter_id ( emitter ) ] ] for emitter in emitters [ : - <NUM_LIT:1> ] ] + [ [ make ( t , emitters [ - <NUM_LIT:1> ] ) for t in self . pending [ self . emitter_id ( emitters [ - <NUM_LIT:1> ] ) ] ] ] <EOL> return dict ( ( k , v ) for k , v in zip ( emitters , result_values ) ) <EOL> def run_simple_topology ( * l , ** kw ) : <EOL> return Mock . run_simple_topology ( * l , ** kw ) </s>
