<s> import logging <EOL> import json <EOL> logger = logging . getLogger ( __name__ ) <EOL> from boto . sqs . message import Message <EOL> from nymms . schemas import Task <EOL> from nymms . probe . Probe import Probe <EOL> from nymms . state . sdb_state import SDBStateManager <EOL> from nymms . utils . aws_helper import SNSTopic , ConnectionManager <EOL> class SQSProbe ( Probe ) : <EOL> def __init__ ( self , region , task_queue , results_topic , state_domain , <EOL> state_manager = SDBStateManager ) : <EOL> self . region = region <EOL> self . queue_name = task_queue <EOL> self . topic_name = results_topic <EOL> self . state_manager = state_manager ( region , state_domain ) <EOL> self . _conn = None <EOL> self . _queue = None <EOL> self . _topic = None <EOL> super ( SQSProbe , self ) . __init__ ( ) <EOL> @ property <EOL> def conn ( self ) : <EOL> if not self . _conn : <EOL> self . _conn = ConnectionManager ( self . region ) <EOL> return self . _conn <EOL> @ property <EOL> def queue ( self ) : <EOL> if not self . _queue : <EOL> self . _queue = self . conn . sqs . create_queue ( self . queue_name ) <EOL> return self . _queue <EOL> @ property <EOL> def topic ( self ) : <EOL> if not self . _topic : <EOL> self . _topic = SNSTopic ( self . region , self . topic_name ) <EOL> return self . _topic <EOL> def get_task ( self , ** kwargs ) : <EOL> wait_time = kwargs . get ( '<STR_LIT>' ) <EOL> timeout = kwargs . get ( '<STR_LIT>' ) + <NUM_LIT:3> <EOL> logger . debug ( "<STR_LIT>" , self . queue_name ) <EOL> task_item = self . queue . read ( visibility_timeout = timeout , <EOL> wait_time_seconds = wait_time ) <EOL> task = None <EOL> if task_item : <EOL> task = Task ( json . loads ( task_item . get_body ( ) ) , origin = task_item ) <EOL> return task <EOL> def resubmit_task ( self , task , delay , ** kwargs ) : <EOL> task . increment_attempt ( ) <EOL> logger . debug ( "<STR_LIT>" , task . id , <EOL> delay ) <EOL> m = Message ( ) <EOL> m . set_body ( json . dumps ( task . serialize ( ) ) ) <EOL> return self . queue . write ( m , delay_seconds = delay ) <EOL> def submit_result ( self , result , ** kwargs ) : <EOL> logger . debug ( "<STR_LIT>" , result . id , <EOL> result . state . name , result . state_type . name ) <EOL> return self . topic . publish ( json . dumps ( result . to_primitive ( ) ) ) <EOL> def delete_task ( self , task ) : <EOL> self . queue . delete_message ( task . _origin ) </s>
