<s> """<STR_LIT>""" <EOL> __author__ = '<STR_LIT>' <EOL> import bson <EOL> import pika <EOL> import pymongo <EOL> import threading <EOL> import time <EOL> class StreamClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uri ) : <EOL> """<STR_LIT>""" <EOL> self . client = pymongo . MongoClient ( uri ) <EOL> def list_streams ( self , uid , token , query = { } ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> res = [ ] <EOL> for obj in collection . find ( query ) : <EOL> res . append ( str ( obj [ '<STR_LIT>' ] ) ) <EOL> return res <EOL> def get_messages ( self , uid , token , interval , iden ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> begin = time . time ( ) - interval <EOL> end = time . time ( ) <EOL> collection = database [ '<STR_LIT>' + str ( iden ) ] <EOL> items = collection . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : begin , "<STR_LIT>" : end } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT:body>" : <NUM_LIT:1> } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:1> } } <EOL> ] ) [ '<STR_LIT:result>' ] <EOL> return list ( items ) <EOL> class AMQPClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uri ) : <EOL> self . client = pymongo . MongoClient ( uri ) <EOL> self . cache = { } <EOL> self . uri = uri <EOL> def info ( self , uid , token ) : <EOL> """<STR_LIT>""" <EOL> res = { } <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> res [ '<STR_LIT>' ] = collection . count ( ) <EOL> return res <EOL> def list_streams ( self , uid , token ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> res = [ ] <EOL> for obj in collection . find ( ) : <EOL> tmp = { '<STR_LIT>' : str ( obj [ '<STR_LIT>' ] ) , '<STR_LIT>' : obj [ '<STR_LIT>' ] } <EOL> if str ( obj [ '<STR_LIT>' ] ) not in self . cache : <EOL> iden = str ( obj [ '<STR_LIT>' ] ) <EOL> content = collection . find_one ( { '<STR_LIT>' : bson . ObjectId ( iden ) } ) <EOL> uri = content [ '<STR_LIT>' ] <EOL> queue = content [ '<STR_LIT>' ] <EOL> self . cache [ iden ] = StreamConsumer ( uid , token , iden , <EOL> self . uri , str ( uri ) , <EOL> str ( queue ) ) <EOL> self . cache [ iden ] . start ( ) <EOL> res . append ( tmp ) <EOL> return res <EOL> def create ( self , uid , token , uri , queue ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> tmp = { '<STR_LIT>' : uri , '<STR_LIT>' : queue , '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] } } <EOL> obj_id = collection . insert ( tmp ) <EOL> return obj_id <EOL> def retrieve ( self , uid , token , iden ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> content = collection . find_one ( { '<STR_LIT>' : bson . ObjectId ( iden ) } ) <EOL> uri = content [ '<STR_LIT>' ] <EOL> queue = content [ '<STR_LIT>' ] <EOL> begin = time . time ( ) - <NUM_LIT> <EOL> end = time . time ( ) <EOL> collection = database [ '<STR_LIT>' + str ( iden ) ] <EOL> items = collection . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : begin , "<STR_LIT>" : end } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT:body>" : <NUM_LIT:1> } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:1> } } <EOL> ] ) [ '<STR_LIT:result>' ] <EOL> return uri , queue , list ( items ) <EOL> def delete ( self , uid , token , iden ) : <EOL> """<STR_LIT>""" <EOL> database = self . client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> collection = database [ '<STR_LIT>' ] <EOL> collection . remove ( { '<STR_LIT>' : bson . ObjectId ( iden ) } ) <EOL> collection = database [ '<STR_LIT>' + str ( iden ) ] <EOL> collection . drop ( ) <EOL> self . cache [ iden ] . stop ( ) <EOL> self . cache . pop ( iden ) <EOL> class StreamConsumer ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uid , token , iden , str_uri , amqp_uri , queue ) : <EOL> super ( StreamConsumer , self ) . __init__ ( ) <EOL> client = pymongo . MongoClient ( str_uri ) <EOL> database = client [ uid ] <EOL> database . authenticate ( uid , token ) <EOL> self . collection = database [ '<STR_LIT>' + str ( iden ) ] <EOL> self . queue = queue <EOL> connection = pika . BlockingConnection ( pika . URLParameters ( amqp_uri ) ) <EOL> self . channel = connection . channel ( ) <EOL> self . channel . queue_declare ( queue = queue ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . channel . basic_consume ( self . callback , queue = self . queue , <EOL> no_ack = True ) <EOL> self . channel . start_consuming ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . channel . stop_consuming ( ) <EOL> def callback ( self , channel , method , properties , body ) : <EOL> """<STR_LIT>""" <EOL> tmp = time . time ( ) <EOL> tmp = { '<STR_LIT>' : tmp , '<STR_LIT:body>' : body } <EOL> self . collection . insert ( tmp ) </s>
