@g.stats.amqp_processor('cloudsearch_changes')
def _run_changed(msgs, chan):
 start = datetime.now(g.tz)
 changed = [pickle.loads(msg.body) for msg in msgs]
 link_fns = LinkUploader.desired_fullnames(changed)
 sr_fns = SubredditUploader.desired_fullnames(changed)
 link_uploader = LinkUploader(g.CLOUDSEARCH_DOC_API, fullnames=link_fns)
 subreddit_uploader = SubredditUploader(g.CLOUDSEARCH_SUBREDDIT_DOC_API, fullnames=sr_fns)
 link_time = link_uploader.inject()
 subreddit_time = subreddit_uploader.inject()
 cloudsearch_time = (link_time + subreddit_time)
 totaltime = (datetime.now(g.tz) - start).total_seconds()
 print ('%s: %d messages in %.2fs seconds (%.2fs secs waiting on cloudsearch); %d duplicates, %s remaining)' % (start, len(changed), totaltime, cloudsearch_time, (len(changed) - len((link_fns | sr_fns))), msgs[(-1)].delivery_info.get('message_count', 'unknown')))

@ g . stats . amqp_processor ( 'cloudsearch_changes' ) def _run_changed ( msgs , chan ) : start = datetime . now ( g . tz ) changed = [ pickle . loads ( msg . body ) for msg in msgs ] link_fns = LinkUploader . desired_fullnames ( changed ) sr_fns = SubredditUploader . desired_fullnames ( changed ) link_uploader = LinkUploader ( g . CLOUDSEARCH_DOC_API , fullnames = link_fns ) subreddit_uploader = SubredditUploader ( g . CLOUDSEARCH_SUBREDDIT_DOC_API , fullnames = sr_fns ) link_time = link_uploader . inject ( ) subreddit_time = subreddit_uploader . inject ( ) cloudsearch_time = ( link_time + subreddit_time ) totaltime = ( datetime . now ( g . tz ) - start ) . total_seconds ( ) print ( '%s: %d messages in %.2fs seconds (%.2fs secs waiting on cloudsearch); %d duplicates, %s remaining)' % ( start , len ( changed ) , totaltime , cloudsearch_time , ( len ( changed ) - len ( ( link_fns | sr_fns ) ) ) , msgs [ ( - 1 ) ] . delivery_info . get ( 'message_count' , 'unknown' ) ) )
consume the cloudsearch_changes queue .