<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import logging <EOL> from pyleus . storm import StormWentAwayError <EOL> from pyleus . storm . component import Component <EOL> log = logging . getLogger ( __name__ ) <EOL> class Spout ( Component ) : <EOL> """<STR_LIT>""" <EOL> COMPONENT_TYPE = "<STR_LIT>" <EOL> def next_tuple ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def ack ( self , tup_id ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def fail ( self , tup_id ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def _handle_command ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> command = msg [ '<STR_LIT>' ] <EOL> if command == '<STR_LIT>' : <EOL> self . next_tuple ( ) <EOL> elif command == '<STR_LIT>' : <EOL> self . ack ( msg [ '<STR_LIT:id>' ] ) <EOL> elif command == '<STR_LIT>' : <EOL> self . fail ( msg [ '<STR_LIT:id>' ] ) <EOL> def _sync ( self ) : <EOL> """<STR_LIT>""" <EOL> self . send_command ( '<STR_LIT>' ) <EOL> def run_component ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> while True : <EOL> msg = self . read_command ( ) <EOL> self . _handle_command ( msg ) <EOL> self . _sync ( ) <EOL> except StormWentAwayError : <EOL> log . warning ( "<STR_LIT>" ) <EOL> def emit ( <EOL> self , values , <EOL> stream = None , tup_id = None , <EOL> direct_task = None , need_task_ids = True ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( values , list ) or isinstance ( values , tuple ) <EOL> command_dict = { <EOL> '<STR_LIT>' : tuple ( values ) , <EOL> } <EOL> if stream is not None : <EOL> command_dict [ '<STR_LIT>' ] = stream <EOL> if tup_id is not None : <EOL> command_dict [ '<STR_LIT:id>' ] = tup_id <EOL> if direct_task is not None : <EOL> command_dict [ '<STR_LIT>' ] = direct_task <EOL> if not need_task_ids : <EOL> command_dict [ '<STR_LIT>' ] = False <EOL> self . send_command ( '<STR_LIT>' , command_dict ) <EOL> if need_task_ids : <EOL> return self . read_taskid ( ) </s>
