<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from collections import defaultdict <EOL> from pystorm . component import Component <EOL> from six import add_metaclass , iteritems , itervalues , string_types <EOL> from thriftpy . transport import TMemoryBuffer <EOL> from thriftpy . protocol import TBinaryProtocol <EOL> from . . thrift import storm_thrift <EOL> from . bolt import JavaBoltSpec , ShellBoltSpec <EOL> from . component import ComponentSpec , ShellComponentSpec <EOL> from . spout import JavaSpoutSpec , ShellSpoutSpec <EOL> from . util import to_python_arg_list <EOL> class TopologyType ( type ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( mcs , classname , bases , class_dict ) : <EOL> bolt_specs = { } <EOL> spout_specs = { } <EOL> specs = TopologyType . class_dict_to_specs ( class_dict ) <EOL> for spec in itervalues ( specs ) : <EOL> if isinstance ( spec , ( JavaBoltSpec , ShellBoltSpec ) ) : <EOL> TopologyType . add_bolt_spec ( spec , bolt_specs ) <EOL> elif isinstance ( spec , ( JavaSpoutSpec , ShellSpoutSpec ) ) : <EOL> TopologyType . add_spout_spec ( spec , spout_specs ) <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( spec ) ) <EOL> TopologyType . clean_spec_inputs ( spec , specs ) <EOL> if classname != '<STR_LIT>' and not spout_specs : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> class_dict [ '<STR_LIT>' ] = bolt_specs <EOL> class_dict [ '<STR_LIT>' ] = spout_specs <EOL> class_dict [ '<STR_LIT>' ] = list ( specs . values ( ) ) <EOL> class_dict [ '<STR_LIT>' ] = storm_thrift . StormTopology ( spouts = spout_specs , <EOL> bolts = bolt_specs , <EOL> state_spouts = { } ) <EOL> return type . __new__ ( mcs , classname , bases , class_dict ) <EOL> @ classmethod <EOL> def class_dict_to_specs ( mcs , class_dict ) : <EOL> """<STR_LIT>""" <EOL> specs = { } <EOL> for name , spec in iteritems ( class_dict ) : <EOL> if isinstance ( spec , ComponentSpec ) : <EOL> if spec . name is None : <EOL> spec . name = name <EOL> if spec . name in specs : <EOL> raise ValueError ( "<STR_LIT>" <EOL> . format ( spec . name ) ) <EOL> else : <EOL> specs [ spec . name ] = spec <EOL> elif isinstance ( spec , Component ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( spec ) ) <EOL> return specs <EOL> @ classmethod <EOL> def add_bolt_spec ( mcs , spec , bolt_specs ) : <EOL> """<STR_LIT>""" <EOL> if not spec . inputs : <EOL> cls_name = spec . component_cls . __name__ <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( cls_name , spec . name ) ) <EOL> bolt_specs [ spec . name ] = storm_thrift . Bolt ( bolt_object = spec . component_object , <EOL> common = spec . common ) <EOL> @ classmethod <EOL> def add_spout_spec ( mcs , spec , spout_specs ) : <EOL> """<STR_LIT>""" <EOL> if not spec . outputs : <EOL> cls_name = spec . component_cls . __name__ <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( cls_name , spec . name ) ) <EOL> spout_specs [ spec . name ] = storm_thrift . SpoutSpec ( spout_object = spec . component_object , <EOL> common = spec . common ) <EOL> @ classmethod <EOL> def clean_spec_inputs ( mcs , spec , specs ) : <EOL> """<STR_LIT>""" <EOL> if spec . inputs is None : <EOL> spec . inputs = { } <EOL> for stream_id , grouping in list ( iteritems ( spec . inputs ) ) : <EOL> if isinstance ( stream_id . componentId , ComponentSpec ) : <EOL> del spec . inputs [ stream_id ] <EOL> stream_id . componentId = stream_id . componentId . name <EOL> spec . inputs [ stream_id ] = grouping <EOL> elif stream_id . componentId is None : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> stream_comp = specs [ stream_id . componentId ] <EOL> valid_fields = set ( stream_comp . outputs [ stream_id . streamId ] <EOL> . output_fields ) <EOL> if grouping . fields is not None : <EOL> for field in grouping . fields : <EOL> if field not in valid_fields : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( field , <EOL> stream_comp . name , <EOL> stream_id . streamId ) ) <EOL> def __repr__ ( cls ) : <EOL> """<STR_LIT>""" <EOL> return repr ( cls . _topology ) <EOL> @ add_metaclass ( TopologyType ) <EOL> class Topology ( object ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def write ( cls , stream ) : <EOL> """<STR_LIT>""" <EOL> def write_it ( stream ) : <EOL> transport_out = TMemoryBuffer ( ) <EOL> protocol_out = TBinaryProtocol ( transport_out ) <EOL> cls . _topology . write ( protocol_out ) <EOL> transport_bytes = transport_out . getvalue ( ) <EOL> stream . write ( transport_bytes ) <EOL> if isinstance ( stream , string_types ) : <EOL> with open ( stream , '<STR_LIT:wb>' ) as output_file : <EOL> write_it ( output_file ) <EOL> else : <EOL> write_it ( stream ) <EOL> @ classmethod <EOL> def read ( cls , stream ) : <EOL> """<STR_LIT>""" <EOL> def read_it ( stream ) : <EOL> stream_bytes = stream . read ( ) <EOL> transport_in = TMemoryBuffer ( stream_bytes ) <EOL> protocol_in = TBinaryProtocol ( transport_in ) <EOL> topology = storm_thrift . StormTopology ( ) <EOL> topology . read ( protocol_in ) <EOL> cls . _topology = topology <EOL> cls . thrift_bolts = topology . bolts <EOL> cls . thrift_spouts = topology . spouts <EOL> cls . specs = [ ] <EOL> if isinstance ( stream , string_types ) : <EOL> with open ( stream , '<STR_LIT:rb>' ) as input_file : <EOL> return read_it ( input_file ) <EOL> else : <EOL> return read_it ( stream ) <EOL> @ staticmethod <EOL> def _spec_to_flux_dict ( spec ) : <EOL> """<STR_LIT>""" <EOL> flux_dict = { '<STR_LIT:id>' : spec . name , <EOL> '<STR_LIT>' : [ ] } <EOL> if isinstance ( spec , ShellComponentSpec ) : <EOL> if isinstance ( spec , ShellBoltSpec ) : <EOL> flux_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> flux_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> shell_object = spec . component_object . shell <EOL> flux_dict [ '<STR_LIT>' ] . append ( [ shell_object . execution_command , <EOL> shell_object . script ] ) <EOL> output_fields = [ '<STR_LIT>' ] <EOL> if '<STR_LIT:default>' in spec . outputs : <EOL> output_fields = spec . outputs [ '<STR_LIT:default>' ] . output_fields <EOL> flux_dict [ '<STR_LIT>' ] . append ( output_fields ) <EOL> if set ( spec . outputs . keys ( ) ) - { '<STR_LIT:default>' } : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( spec ) ) <EOL> else : <EOL> if spec . component_object . serialized_java is not None : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( spec ) ) <EOL> java_object = spec . component_object . java_object <EOL> flux_dict [ '<STR_LIT>' ] = java_object . full_class_name <EOL> flux_dict [ '<STR_LIT>' ] = to_python_arg_list ( java_object . args_list ) <EOL> return flux_dict <EOL> @ staticmethod <EOL> def _stream_to_flux_dict ( spec , global_stream , grouping ) : <EOL> """<STR_LIT>""" <EOL> flux_dict = { '<STR_LIT>' : global_stream . componentId , <EOL> '<STR_LIT:to>' : spec . name } <EOL> grouping_dict = { '<STR_LIT>' : global_stream . streamId } <EOL> for key , val in grouping . __dict__ . items ( ) : <EOL> if val is not None : <EOL> grouping_dict [ '<STR_LIT:type>' ] = key . upper ( ) <EOL> if key == '<STR_LIT>' : <EOL> if val : <EOL> grouping_dict [ '<STR_LIT:args>' ] = val <EOL> else : <EOL> grouping_dict [ '<STR_LIT:type>' ] = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> grouping_dict [ '<STR_LIT:type>' ] = '<STR_LIT>' <EOL> class_dict = { '<STR_LIT>' : val . full_class_name , <EOL> '<STR_LIT:args>' : to_python_arg_list ( val . arg_list ) } <EOL> grouping_dict [ '<STR_LIT>' ] = class_dict <EOL> flux_dict [ '<STR_LIT>' ] = grouping_dict <EOL> return flux_dict <EOL> @ classmethod <EOL> def to_flux_dict ( cls , name ) : <EOL> """<STR_LIT>""" <EOL> flux_dict = { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] } <EOL> for spec in cls . specs : <EOL> if isinstance ( spec , ( JavaBoltSpec , ShellBoltSpec ) ) : <EOL> flux_dict [ '<STR_LIT>' ] . append ( cls . _spec_to_flux_dict ( spec ) ) <EOL> for global_stream , grouping in spec . inputs . items ( ) : <EOL> stream_dict = cls . _stream_to_flux_dict ( spec , <EOL> global_stream , <EOL> grouping ) <EOL> flux_dict [ '<STR_LIT>' ] . append ( stream_dict ) <EOL> elif isinstance ( spec , ( JavaSpoutSpec , ShellSpoutSpec ) ) : <EOL> flux_dict [ '<STR_LIT>' ] . append ( cls . _spec_to_flux_dict ( spec ) ) <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( spec ) ) <EOL> flux_dict = { key : val for key , val in flux_dict . items ( ) if val } <EOL> return flux_dict </s>
