<s> """<STR_LIT>""" <EOL> import logging <EOL> try : <EOL> import queue <EOL> except : <EOL> import Queue as queue <EOL> import sleekxmpp <EOL> from sleekxmpp . plugins . base import base_plugin <EOL> log = logging . getLogger ( __name__ ) <EOL> class kestrel_client ( base_plugin ) : <EOL> def plugin_init ( self ) : <EOL> self . description = "<STR_LIT>" <EOL> self . submit_jid = self . config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . pool_jid = self . config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . submit_queue = queue . Queue ( ) <EOL> self . status_queue = queue . Queue ( ) <EOL> self . cancel_queue = queue . Queue ( ) <EOL> self . timeout = <NUM_LIT:4> * self . xmpp . response_timeout <EOL> def submit_job ( self , job ) : <EOL> reqs = job . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if isinstance ( reqs , str ) : <EOL> reqs = reqs . upper ( ) <EOL> reqs = reqs . split ( ) <EOL> reqs . sort ( ) <EOL> log . debug ( "<STR_LIT>" % ( self . submit_jid , job ) ) <EOL> self . xmpp [ '<STR_LIT>' ] . prep_handlers ( [ self . _submit_next , <EOL> self . _submit_error , <EOL> self . _submit_complete ] ) <EOL> session = { <EOL> '<STR_LIT>' : job , <EOL> '<STR_LIT>' : reqs , <EOL> '<STR_LIT>' : self . _submit_next , <EOL> '<STR_LIT:error>' : self . _submit_error <EOL> } <EOL> self . xmpp [ '<STR_LIT>' ] . start_command ( self . submit_jid , <EOL> '<STR_LIT>' , <EOL> session ) <EOL> try : <EOL> result = self . submit_queue . get ( block = False , <EOL> timeout = self . timeout ) <EOL> except : <EOL> result = False <EOL> return result <EOL> def _submit_next ( self , iq , session ) : <EOL> job = session [ '<STR_LIT>' ] <EOL> reqs = session [ '<STR_LIT>' ] <EOL> form = self . xmpp [ '<STR_LIT>' ] . makeForm ( ftype = '<STR_LIT>' ) <EOL> form . addField ( var = '<STR_LIT>' , value = job [ '<STR_LIT>' ] ) <EOL> form . addField ( var = '<STR_LIT>' , value = job . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> form . addField ( var = '<STR_LIT>' , value = job . get ( '<STR_LIT>' , '<STR_LIT:1>' ) ) <EOL> form . addField ( var = '<STR_LIT>' , ftype = '<STR_LIT>' , <EOL> value = "<STR_LIT:\n>" . join ( reqs ) ) <EOL> session [ '<STR_LIT>' ] = form <EOL> session [ '<STR_LIT>' ] = self . _submit_complete <EOL> self . xmpp [ '<STR_LIT>' ] . continue_command ( session ) <EOL> def _submit_complete ( self , iq , session ) : <EOL> form = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> job_id = form [ '<STR_LIT>' ] . get ( '<STR_LIT>' , False ) <EOL> self . submit_queue . put ( job_id ) <EOL> session [ '<STR_LIT>' ] = None <EOL> session [ '<STR_LIT>' ] = None <EOL> self . xmpp [ '<STR_LIT>' ] . complete_command ( session ) <EOL> def _submit_error ( self , iq , session ) : <EOL> self . submit_queue . put ( False ) <EOL> error = iq [ '<STR_LIT:error>' ] [ '<STR_LIT>' ] <EOL> log . error ( "<STR_LIT>" % error ) <EOL> def cancel_job ( self , job_id ) : <EOL> return self . cancel_jobs ( set ( ( job_id , ) ) ) <EOL> def cancel_jobs ( self , job_ids ) : <EOL> self . xmpp [ '<STR_LIT>' ] . prep_handlers ( [ self . _cancel_next , <EOL> self . _cancel_complete , <EOL> self . _cancel_error ] ) <EOL> session = { '<STR_LIT>' : job_ids , <EOL> '<STR_LIT>' : self . _cancel_next , <EOL> '<STR_LIT:error>' : self . _cancel_error } <EOL> self . xmpp [ '<STR_LIT>' ] . start_command ( self . submit_jid , <EOL> '<STR_LIT>' , <EOL> session ) <EOL> try : <EOL> result = self . cancel_queue . get ( block = True , <EOL> timeout = self . timeout ) <EOL> except : <EOL> result = False <EOL> return result <EOL> def _cancel_next ( self , iq , session ) : <EOL> form = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> jobs = form [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> job_ids = session [ '<STR_LIT>' ] <EOL> user_jobs = set ( [ o [ '<STR_LIT:value>' ] for o in jobs . getOptions ( ) ] ) <EOL> jobs = user_jobs . intersection ( job_ids ) <EOL> if jobs : <EOL> form = self . xmpp [ '<STR_LIT>' ] . makeForm ( ftype = '<STR_LIT>' ) <EOL> form . addField ( ftype = '<STR_LIT>' , var = '<STR_LIT>' ) <EOL> form [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:value>' ] = list ( jobs ) <EOL> session [ '<STR_LIT>' ] = form <EOL> session [ '<STR_LIT>' ] = self . _cancel_complete <EOL> self . xmpp [ '<STR_LIT>' ] . complete_command ( session ) <EOL> else : <EOL> log . error ( '<STR_LIT>' ) <EOL> self . cancel_queue . put ( False ) <EOL> session [ '<STR_LIT>' ] = None <EOL> session [ '<STR_LIT>' ] = None <EOL> self . xmpp [ '<STR_LIT>' ] . cancel_command ( session ) <EOL> def _cancel_complete ( self , iq , session ) : <EOL> log . info ( '<STR_LIT>' % list ( session [ '<STR_LIT>' ] ) ) <EOL> self . cancel_queue . put ( True ) <EOL> def _cancel_error ( self , iq , session ) : <EOL> self . cancel_queue . put ( False ) <EOL> error = iq [ '<STR_LIT:error>' ] [ '<STR_LIT>' ] <EOL> log . error ( '<STR_LIT>' % error ) <EOL> def pool_status ( self ) : <EOL> iq = self . xmpp [ '<STR_LIT>' ] . send_command ( self . pool_jid , <EOL> '<STR_LIT>' ) <EOL> if iq is not None and iq [ '<STR_LIT:type>' ] != '<STR_LIT:error>' : <EOL> session = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> form = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return form [ '<STR_LIT>' ] <EOL> log . error ( "<STR_LIT>" ) <EOL> return False <EOL> def job_status ( self , job_id = None ) : <EOL> iq = self . xmpp [ '<STR_LIT>' ] . send_command ( self . submit_jid , <EOL> '<STR_LIT>' ) <EOL> if iq is not None and iq [ '<STR_LIT:type>' ] != '<STR_LIT:error>' : <EOL> session = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> form = iq [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return form [ '<STR_LIT>' ] <EOL> log . error ( "<STR_LIT>" ) <EOL> return False </s>
