def stop(self):
     env = C.get('env');
     host = C.get('rds_host_' + env)
     db = C.get('rds_db_' + env)
     rds = Redis(host = host, port = 6379, db = db)
     tradeChannel = C.get('channel_trade')
     rds.publish(tradeChannel, 'stop');

class Subscriber(object):
    def __init__(self):
        self.redis = Redis(**settings.REDIS)
        self.events = {}

    def wait(self, game_id):
        if not self.events:
            eventlet.spawn_n(self._listen)

        if game_id not in self.events:
            self.events[game_id] = eventlet.event.Event()
            self.redis.subscribe('scrabble.%d' % game_id)

        # XXX we need to clean up events, otherwise the size of this dict will
        # be unbounded.
        return self.events[game_id].wait()

    def _listen(self):
        # We need to be subscribed to something before redis.listen will work.
        while not self.redis.subscribed:
            eventlet.sleep(1)

        for msg in self.redis.listen():
            if msg['type'] == 'message':
                _, game_id = msg['channel'].split('.')
                game_id = int(game_id)
                msg_type, player_num, msg = msg['data'].split(':', 2)
                self.events[game_id].send()
                self.events[game_id].reset()

 def _get_apikey(self):
     redis = Redis(**config.REDIS_INFO)
     for apikey in redis.keys('*'):
         owner = redis.get(apikey).decode()
         if self.name == owner:
             return apikey
     return None

def create_datasets_list():
    if USE_REDIS:
        key = "all_datasets"
        result = Redis.get(key)

        if result:
            logger.debug("Redis cache hit")
            datasets = pickle.loads(result)

    if result is None:
        datasets = list()
        with Bench("Creating DataSets object"):
            type_dict = {'Publish': 'PublishFreeze',
                         'ProbeSet': 'ProbeSetFreeze',
                         'Geno': 'GenoFreeze'}

            for dataset_type in type_dict:
                query = "SELECT Name FROM {}".format(type_dict[dataset_type])
                for result in fetchall(query):
                    #The query at the beginning of this function isn't
                    #necessary here, but still would rather just reuse
                    #it logger.debug("type: {}\tname:
                    #{}".format(dataset_type, result.Name))
                    dataset = create_dataset(result.Name, dataset_type)
                    datasets.append(dataset)

        if USE_REDIS:
            Redis.set(key, pickle.dumps(datasets, pickle.HIGHEST_PROTOCOL))
            Redis.expire(key, 60*60)

    return datasets

def group_app_task_out(lock, name, appops, timeout):
    from redis import Redis
    redis = Redis()
    try:
        group_app_task(redis, lock, name, appops, timeout)
    finally:
        redis.delete(lock)

class CloudAggregator:
    """
    This class is responsible for querying remote Cloud sites, retrieving their resource and real time XML,
    aggregate and validate the retrieved XML and then finally storing the aggregated XML into a RedisDB
    """
    # Since this is the "primary" class, it is designed to be instantiated first and thus will load the 
    # global ConfigMapping data
    def __init__(self, logger=None):
        if(logger):
            self.logger = logger
        else:
            self.logger = Logger("cloud_aggregator", "cloud_aggregator.log")
        loadConfig(self.logger) 
        
        #Connect to the RedisDB with the configured options
        self.storageDb = Redis(db=ConfigMapping[TARGET_REDIS_DB], host=ConfigMapping[REDISDB_SERVER_HOSTNAME], port=int(ConfigMapping[REDISDB_SERVER_PORT]))

        # Verify the DB is up and running
        try:
           self.storageDb.ping()
           self.logger.debug("RedisDB server alive")
        except ConnectionError, err:
#            print str(err)
            self.logger.error("redis-server running on desired port? "+str(err))
            sys.exit(RET_CRITICAL)

def send_mail_all():
    """
    God Bless me that can me read this code after a month!
    """
    r = Redis()
    group_levels = []
    while 1:
        group_level = r.spop('group_level_set')
        if group_level == None:
            break
        group_levels.append(group_level)
    print('{0} are to be processed.'.format(group_levels))
    for g_l in group_levels:
        # we do send mail here by group/level pair
        group,level = g_l.split('_')
        mailto_list = get_sendlist_by_group_level(int(group),int(level))
        # if nobody wants this log
        if len(mailto_list) == 0:
            continue
        print('group:{0} level:{1} mailto:{2} is to be sent!'.format(group,level,mailto_list))
        content_list = []
        while 1:
            content = r.lpop(g_l)
            if content == None:
                break
            content_list.append(content)
        body = render_body(content_list)
        SM.SendMail(content=body,mailto=mailto_list)

def get_status_message(conn: redis.Redis, uid, timeline='home:', page=1, count=30):
    statuses = conn.zrevrange('%s%s' % (timeline, uid), (page - 1) * count, page * count - 1)
    pipeline = conn.pipeline(True)
    for id in statuses:
        pipeline.hgetall('status:%s' % id)
    # 使用 filter 过滤已经被删除的消息
    return filter(None, pipeline.execute())

def run(architecture_file, model_file, num_batches=10, pattern="centralModel-*"):
    print "evaluating Q values..."
    redisInstance = Redis(host='localhost', port=6379, db=0)
    model_keys = redisInstance.keys(pattern)
    results = {}

    net = BaristaNet(architecture_file, model_file, None)
    replay_dataset = ReplayDataset("temp-q-converge-dset.hdf5",
                                   net.state[0].shape,
                                   dset_size=1000,
                                   overwrite=True)
    net.add_dataset(replay_dataset)

    game = SnakeGame()
    preprocessor = generate_preprocessor(net.state.shape[2:], gray_scale)
    exp_gain = ExpGain(net, ['w', 'a', 's', 'd'], preprocessor, game.cpu_play,
                       replay_dataset, game.encode_state())

    print "Generating new experiences..."
    for _ in xrange(100):
        exp_gain.generate_experience(1e5)
    print "Done"

    for key in model_keys:
        print "Evaluating model:", key
        model = dict(redisC.Dict(key=key, redis=redisInstance))
        q_avg = evaluate_model(net, model, num_batches)
        results[key] = q_avg

    for key in sorted(results.keys()):
        print key.ljust(25) + "%0.4f" % results[key]

class DatamartApplication(Application):

    def __init__(self, **kwargs):
        routes = [
            (r'/socket', DatamartHandler),
            (r'/(?P<model>task|sprint|user)/(?P<pk>[0-9]+)', UpdateHandler),
            (r'/websocket', EchoHandler),
            
        ]
        super().__init__(routes, **kwargs)
        self.subscriber = RedisSubscriber(Client())
        self.publisher = Redis()
        self._key = os.environ.get('WATERCOOLER_SECRET',
            'pTyz1dzMeVUGrb0Su4QXsP984qTlvQRHpFnnlHuH')
        self.signer = TimestampSigner(self._key)

    def add_subscriber(self, channel, subscriber):
        self.subscriber.subscribe(['all', channel], subscriber)

    def remove_subscriber(self, channel, subscriber):
        self.subscriber.unsubscribe(channel, subscriber)
        self.subscriber.unsubscribe('all', subscriber)

    def broadcast(self, message, channel=None, sender=None):
        channel = 'all' if channel is None else channel
        message = json.dumps({
            'sender': sender and sender.uid,
            'message': message
        })
        self.publisher.publish(channel, message)

def create_user(conn: redis.Redis, login: str, name):
    llogin = login.lower()
    # 加锁防止多个请求在同一时间使用相同用户名创建新用户
    lock = acquire_lock_with_timeout(conn, 'user:' + llogin, 1)
    if not lock:
        return None
    # `users:` 中存储了小写用户名与 id 之间的映射关系
    if conn.hget('users:', llogin):
        # 说明用户已注册
        release_lock(conn, 'user:' + llogin, lock)
        return None
    id = conn.incr('user:id:')
    pipeline = conn.pipeline(True)
    pipeline.hset('users:', llogin, id)
    pipeline.hmset('user:%s:' % id, {
        'login': login,
        'id': id,
        'name': name,
        'followers': 0,
        'following': 0,
        'posts': 0,
        'signup': time.time()
    })
    pipeline.execute()
    release_lock(conn, 'user:' + llogin, lock)
    return id

class ScrumApplication(Application):

    def __init__(self, **kwargs):
        routes = [
         (r'/socket', SprintHandler),
         (r'/(?P<model>task|sprint|user)/(?P<pk>[0-9]+)', UpdateHandler),
        ]
        super().__init__(routes, **kwargs)
        self.subscriber = RedisSubscriber(Client())
        self.publisher = Redis()
        self._key = os.environ.get('TORNADO_SECRET',
            '[email protected]!d890z103b^4f6k380b+25')
        self.signer = TimestampSigner(self._key)

    def add_subscriber(self, channel, subscriber):
        self.subscriber.subscribe(['all', channel], subscriber)

    def remove_subscriber(self, channel, subscriber):
        self.subscriber.unsubscribe(channel, subscriber)
        self.subscriber.unsubscribe('all', subscriber)

    def broadcast(self, message, channel=None, sender=None):
        channel = 'all' if channel is None else channel
        messsage = json.dumps({
            'sender': sender and sender.uid,
            'message': message
            })
        self.publisher.publish(channel, message)

def start_kuaidi_task():
    r = Redis(host=REDIS_HOST, port=REDIS_PORT)
    kuaidi_key = r.keys("kuaidi:*:auth")
    print kuaidi_key
    if kuaidi_key:
        sid = kuaidi_key[0].split(':')[1]
        loop_kuaidi_task(sid)

    def stream_via_token(cls, token):
        '''
        Set token user to online and publish presence of this user to all
        friends.
        '''
        NereidUser = Pool().get('nereid.user')

        if hasattr(current_app, 'redis_client'):
            redis_client = current_app.redis_client
        else:
            redis_client = Redis(
                CONFIG.get('redis_host', 'localhost'),
                int(CONFIG.get('redis_port', 6379))
            )

        key = 'chat:token:%s' % token
        if not redis_client.exists(key):
            abort(404)

        nereid_user = NereidUser(int(redis_client.get(key)))
        nereid_user.broadcast_presence()

        return Response(
            cls.generate_event_stream(
                nereid_user.id,
                Transaction().cursor.dbname
            ),
            mimetype='text/event-stream'
        )

class RedisQueue(object):
    def __init__(self, crawler):
        try:
            from redis import Redis
        except ImportError:
            raise NotConfigured

        settings = crawler.settings

        # get settings
        queue = settings.get("REDIS_QUEUE")
        if queue is None:
            raise NotConfigured

        host = settings.get("REDIS_HOST", "localhost")
        port = settings.getint("REDIS_PORT", 6379)
        db = settings.getint("REDIS_DB", 0)
        password = settings.get("REDIS_PASSWORD")

        self.redis = Redis(host=host, port=port, db=db, password=password)
        self.queue = queue
        self.project = settings["BOT_NAME"]

        crawler.signals.connect(self.spider_closed, signal=signals.spider_closed)

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler)

    def spider_closed(self, spider, reason):
        msg = {"project": self.project, "spider": spider.name, "reason": reason}
        self.redis.rpush(self.queue, pickle.dumps(msg))

def crawl_tweets_for_event(event_id):
    r = Redis()
    p = HTMLParser()

    total_tweets = 0

    event_title = r.get("festival:%s:title" % event_id).decode("utf-8", errors="ignore")
    event_title = strip_accents(event_title)
    event_title = p.unescape(event_title)
    event_title = remove_stopwords(event_title)

    artists = r.get("festival:%s:artists" % event_id)

    for k, v in eval(artists).items():
        if type(v) == list:
            for artist in v:
                print F, "searching tweets for %s %s" % (k, artist)
                total_tweets += search_term(artist)
        elif type(v) == str:
            print F, "searching tweets for %s %s" % (k, v)
            total_tweets += search_term(v)

    r.incr("festival:%s:crawled_times" % event_id)

    print F, "searching tweets for festival title: %s" % event_title
    total_tweets += search_term(event_title, event_id)  # newsid
    print F, "total tweets: %d" % total_tweets

class Redis:
    def __init__(self, key_prefix, key_sufix):
        self.key_prefix = key_prefix
        self.key_sufix = key_sufix
        self.host = settings.OPPS_DB_HOST
        self.port = settings.OPPS_DB_PORT
        self.db = 0

        pool = ConnectionPool(host=self.host,
                              port=self.port,
                              db=self.db)
        self.conn = RedisClient(connection_pool=pool)

    def object(self):
        return self.conn

    def close(self):
        self.conn = None
        return True

    @property
    def key(self):
        return u'{}_{}_{}'.format(settings.OPPS_DB_NAME,
                                  self.key_prefix,
                                  self.key_sufix).lower()

    def save(self, document):
        return self.conn.set(self.key, document)

    def publish(self, document):
        return self.conn.publish(self.key, document)

    def get(self):
        return self.conn.get(self.key)

def redis(request):
  redis = Redis()
  redis.flushdb()
  def finalizer():
    redis.delete(TEST_QUEUE_NAME)
  request.addfinalizer(finalizer)
  return redis

class RedisConnector(object):
    ROBOT_LIBRARY_SCOPE = 'GLOBAL'

    logger = RedisLogger('RedisConnector')
    pool = ConnectionPool(host='localhost', port=6379, db=0, socket_timeout=5)

    def __init__(self):
        self.redis = Redis(connection_pool=RedisConnector.pool)

    def check_key_exists(self, **ka):
        """
        Key existence True/1 or False/0
        """
        return self.redis.exists(name=ka['key'])

    @checker
    def get_key_field(self, **ka):
        return self.redis.hget(name=ka['key'], key=ka['field'])

    @checker
    def get_key_fields(self, **ka):
        return self.redis.hgetall(name=ka['key'])

    @checker
    def get_key_ttl(self, **ka):
        return self.redis.ttl(name=ka['key'])

    def set_key_field(self, **ka):
        return self.redis.hset(name=ka['key'], key=ka['field'], value=ka['value'])

class Database:

    def __init__(self, db_num=0):
        self.redis = Redis(db=db_num)

    @staticmethod
    def concat(*args):
        return '.'.join([str(a) for a in args])

    def get_blocks(self, blocks):
        '''
        :param blocks: list of hashes
        :return: list of Block objects (encodium)
        '''
        return [SimpleBlock.from_json(self.redis.get(str(block_hash)).decode()) for block_hash in blocks]

    def set_kv(self, key, value):
        return self.redis.set(key, serialize_if_encodium(value))

    def get_kv(self, key, optional_type=None):
        result = self.redis.get(key)
        print('getting', key, result)
        if optional_type is not None and result is not None:
            if issubclass(optional_type, Encodium):
                return optional_type.from_json(result.decode())
            return optional_type(result)
        return result