class RedisProducer(object):
    def __init__(self, hostname = 'localhost', port = 6379):
        log.debug("Initializing RedisProducer with hostname of %s and port %s" % (hostname, port))
        self.r = StrictRedis(host = hostname, port = port)

    def send(self, message):
        tries = 0
        next_index_key = get_next_index_for_topic_key(message.topic)
        next_index = 1
        result = None
        log.debug("Sending message on topic %s" % message.topic)

        while result is None and tries < TRIES_LIMIT:
            if self.r.exists(next_index_key):
                next_index = long(self.r.get(next_index_key)) + 1

            message_key = get_message_key(message.topic, next_index)

            try:
                pl = self.r.pipeline()
                pl.watch(next_index_key, message_key)
                pl.multi()
                pl.incr(next_index_key).set(message_key, message.payload)
                result = pl.execute()
            except WatchError:
                # Should probably log something here, but all it means is we're
                # retrying
                pass

        if result is None:
            log.error("Could not send message, retry amount exceeded")
            raise RuntimeError("Attempted to send message %s times and failed" % TRIES_LIMIT)

 def sync_get(self, identity, *args, **kwargs):
     """
     For getting data from cache
     :param identity: Unique Integer for the data
     :param args: Args for the sync function. (Default: None)
     """
     redis = StrictRedis(connection_pool=self.redis_pool)
     key = key_generator(self.key, identity)
     try:
         if redis.exists(key):
             data = self.get_func(redis.get(key))
         else:
             data = self.sync_func(identity, *args, **kwargs)
             if self.expire:
                 self._setex(redis, key, self.set_func(data))
             else:
                 redis.set(key, self.set_func(data))
         if data is not None or data != "":
             return data
         return None
     except RedisError as re:
         self.log.error("[REDIS] %s", str(re))
         data = self.sync_func(identity, args)
         return data
     finally:
         del redis

def load_data(ids, host, port):
    """from redis
    uniq in index
    """
    from redis import StrictRedis
    from json import loads

    db = StrictRedis(host, port, decode_responses=True)
    logging.debug("all: %d", len(ids))

    if 0 in ids:
        raise KeyError(0)

    from collections import Counter
    checker = Counter(ids)
    if len(checker) != len(ids):  # not unique
        raise ValueError(checker.most_common(3))

    pipe = db.pipeline()
    ids.insert(0, 0)
    for i in ids:
        pipe.hgetall(i)
    properties = pipe.execute(True)  # DO NOT allow error occurs in redis

    raw = {}
    for i, p in zip(ids, properties):
        raw[i] = {k: loads(v) for k, v in p.items()}
    return raw

 def redis(self):
     redis_config = self.config.get('redis')
     redis = StrictRedis(
         redis_config.get('host'), redis_config.get('port'), redis_config.get('db')
     )
     redis.ping()
     return redis

class SutroTestCase(TestCase):
  def setUp(self):
    redis_connection_pool = ConnectionPool(**settings.WS4REDIS_CONNECTION)
    self.redis = StrictRedis(connection_pool=redis_connection_pool)
    self.client = Client()

  def tearDown(self):
    self.redis.flushdb()

  def random_string(self, length=None, str_type=None):
    DEFAULT_LENGTH = 10
    length = length if length else DEFAULT_LENGTH
    if str_type == 'number':
      string_type = string.digits
    else:
      string_type = string.lowercase
    return ''.join(random.choice(string_type) for x in range(length))

  def create_a_user(self, display_name, user_id=None, icon_url=None, user_url=None, rdio_key=None):
    user = User()
    user.display_name = display_name
    user.id = user_id if user_id else uuid.uuid4().hex
    user.icon_url = icon_url if icon_url else 'http://' + self.random_string() + '.jpg'
    user.user_url = user_url if user_url else 'http://' + self.random_string() + '.com/' + self.random_string()
    user.rdio_key = rdio_key if rdio_key else 's' + self.random_string(length=5, str_type='number')
    user.save(self.redis)
    return user

  def create_a_party(self, party_id, name):
    party = Party()
    party.id = party_id
    party.name = name
    party.save(self.redis)
    return party

def add_tweets_to_redis(tweet_file):
    """
    'DVD: FBI WARNING Me: oh boy here we go DVD: The board advises you
    to have lots of fun watching this Hollywood movie Me: Ah.. It's a
    nice one'
    """
    redis_client = StrictRedis(host='localhost', port=6379, db=0)
    with open(tweet_file, 'r') as tweets:
        for line in tweets:
            # again, dealing with weird error here
            try:
                tweet = line.strip().split('|', 2)[2]
                # need to investigate whether one-by-one inserting
                # or building a list of tweets and doing a single insert
                # would be more efficient
                if not redis_client.sismember('tweets', tweet):
                    result = redis_client.sadd('tweets', tweet)
                    if not result:
                        print('error occurred adding tweet: "{}" to redis'
                              .format(tweet))
                else:
                    # found all new tweets
                    break
            except IndexError:
                continue
    redis_client.save()

class RedisDataStore(DataStore):
    """Redis-backed datastore object."""

    def __init__(self, number=0):
        redis_host = os.environ.get('REDIS_PORT_6379_TCP_ADDR')
        redis_port = os.environ.get('REDIS_PORT_6379_TCP_PORT')
        self.redis_conn = StrictRedis(host=redis_host, port=redis_port,
                                      db=number)

    def __setitem__(self, k, v):
        self.redis_conn.set(k, v)

    def __getitem__(self, k):
        return self.redis_conn.get(k)

    def __delitem__(self, k):
        self.redis_conn.delete(k)

    def get(self, k):
        return self.redis_conn.get(k)

    def __contains__(self, k):
        return self.redis_conn.exists(k)

    def todict(self):
        #TODO(tvoran): use paginate
        #TODO(tvoran): do something besides multiple gets
        data = {}
        for key in self.redis_conn.keys():
            data[key] = self.get(key)
        return data

    def clear_all(self):
        self.redis_conn.flushdb()

def test_token(redis_server):
    conn = StrictRedis(unix_socket_path=UDS_PATH)
    lock = Lock(conn, "foobar-tok")
    tok = lock.id
    assert conn.get(lock._name) is None
    lock.acquire(blocking=False)
    assert conn.get(lock._name) == tok

def get_redis_info():
    """Check Redis connection."""
    try:
        url = settings.BROKER_URL
        _, host, port, _, password, db, _ = parse_redis_url(url)
    except AttributeError:
        log.error("No valid Redis connection info found in settings.")
        return {"status": NO_CONFIG}

    start = datetime.now()
    try:
        rdb = StrictRedis(
            host=host, port=port, db=db,
            password=password, socket_timeout=TIMEOUT_SECONDS,
        )
        info = rdb.info()
    except (RedisConnectionError, TypeError) as ex:
        log.error("Error making Redis connection: %s", ex.args)
        return {"status": DOWN}
    except RedisResponseError as ex:
        log.error("Bad Redis response: %s", ex.args)
        return {"status": DOWN, "message": "auth error"}
    micro = (datetime.now() - start).microseconds
    del rdb  # the redis package does not support Redis's QUIT.
    ret = {
        "status": UP, "response_microseconds": micro,
    }
    fields = ("uptime_in_seconds", "used_memory", "used_memory_peak")
    ret.update({x: info[x] for x in fields})
    return ret

def test_lshash_redis_extra_val():
    """
    Test external lshash module
    """
    config = {"redis": {"host": 'localhost', "port": 6379, "db": 15}}
    sr = StrictRedis(**config['redis'])
    sr.flushdb()

    lsh = LSHash(6, 8, 1, config)
    for i in xrange(num_elements):
        lsh.index(list(els[i]), el_names[i])
        lsh.index(list(els[i]), el_names[i])  # multiple insertions
    hasht = lsh.hash_tables[0]
    itms = [hasht.get_list(k) for k in hasht.keys()]
    for itm in itms:
        assert itms.count(itm) == 1
        for el in itm:
            assert el[0] in els
            assert el[1] in el_names
    for el in els:
        res = lsh.query(list(el), num_results=1, distance_func='euclidean')[0]
        # vector an name are in the first element of the tuple res[0]
        el_v, el_name = res[0]
        # the distance is in the second element of the tuple
        el_dist = res[1]
        assert el_v in els
        assert el_name in el_names
        assert el_dist == 0
    del lsh
    sr.flushdb()

 def __init_redis(self):
     try:
         redis = StrictRedis()
         redis.ping()    # raises an exception if it failes
         self.redis = redis
     except:
         pass

def test_lshash_redis():
    """
    Test external lshash module
    """
    config = {"redis": {"host": 'localhost', "port": 6379, "db": 15}}
    sr = StrictRedis(**config['redis'])
    sr.flushdb()

    lsh = LSHash(6, 8, 1, config)
    for i in xrange(num_elements):
        lsh.index(list(els[i]))
        lsh.index(list(els[i]))  # multiple insertions should be prevented by the library
    hasht = lsh.hash_tables[0]
    itms = [hasht.get_list(k) for k in hasht.keys()]
    for itm in itms:
        for el in itm:
            assert itms.count(itm) == 1  # have multiple insertions been prevented?
            assert el in els
    for el in els:
        res = lsh.query(list(el), num_results=1, distance_func='euclidean')[0]
        el_v, el_dist = res
        assert el_v in els
        assert el_dist == 0
    del lsh
    sr.flushdb()

  def __init__(self, hosts, options):
    sentinel_kwargs = self._get_sentinel_kwargs(options)
    node_kwargs = self._get_node_kwargs(options)

    masters = None
    # Try to fetch a list of all masters from any sentinel.
    hosts = list(hosts)
    shuffle(hosts) # Randomly sort sentinels before trying to bootstrap.
    for host, port in hosts:
      client = StrictRedis(host=host, port=port, **sentinel_kwargs)
      try:
        masters = client.sentinel_masters().keys()
        break
      except RedisError:
        pass
    if masters is None:
      # No Sentinel responded successfully?
      raise errors.MastersListUnavailable
    if not len(masters):
      # The masters list was empty?
      raise errors.NoMastersConfigured

    sentinel_kwargs.update({
      # Sentinels connected to fewer sentinels than `MIN_SENTINELS` will
      # be ignored.
      'min_other_sentinels': options.get('MIN_SENTINELS',
                                         len(hosts) / 2),
      })
    self.sentinel = Sentinel(hosts, **sentinel_kwargs)
    masters = [self.sentinel.master_for(name, **node_kwargs)
               for name in masters]
    super(SentinelBackedRingClient, self).__init__(masters, options)

def show_search_results(query):
    print("""
Search Result:  {keyword}
==================================================

""".format(keyword=query))

    p = Persistent("minamo")
    r = StrictRedis(decode_responses=True)


    resultset = None
    for _query in query.split(" "):
        qgram = ngram.ngram(_query, 2)
        for bi in list(qgram)[:-1]:
            if resultset is None:
                resultset = set(r.lrange("minamo:bi:{}".format(bi), 0, -1))
            else:
                resultset = resultset & set(r.lrange("minamo:bi:{}".format(bi), 0, -1))

    for page in (p.load(models.Page, x) for x in resultset):
        if page.title is None:
            continue

        print("*", page.title)
        print(" ", page.url)

def check_running(name: str):
    socket_path = get_socket_path(name)
    try:
        r = StrictRedis(unix_socket_path=socket_path)
        return r.ping()
    except ConnectionError:
        return False

class RedisManager(NoSqlManager):
    def __init__(self,
                 namespace,
                 url=None,
                 data_dir=None,
                 lock_dir=None,
                 **params):
        self.db = params.pop('db', None)
        self.dbpass = params.pop('password', None)
        self.connection_pool = params.get('redis_connection_pool', None)
        self.expires = params.get('expires', params.get('expiretime', None))
        NoSqlManager.__init__(self,
                              namespace,
                              url=url,
                              data_dir=data_dir,
                              lock_dir=lock_dir,
                              **params)

    def open_connection(self, host, port, **params):
        if not self.connection_pool:
            self.connection_pool = ConnectionPool(host=host, port=port, db=self.db,
                    password=self.dbpass)
        self.db_conn = StrictRedis(connection_pool=self.connection_pool, **params)
    
    def __contains__(self, key):
        return self.db_conn.exists(self._format_key(key))

    def set_value(self, key, value, expiretime=None):
        key = self._format_key(key)
        # beaker.container.Value.set_value calls NamespaceManager.set_value
        # however it (until version 1.6.4) never sets expiretime param.
        #
        # Checking "type(value) is tuple" is a compromise
        # because Manager class can be instantiated outside container.py (See: session.py)
        if (expiretime is None) and (type(value) is tuple):
            expiretime = value[1]
        # If the machinery above fails, then pickup the expires time from the
        # init params.
        if not expiretime and self.expires is not None:
            expiretime = self.expires
        # Set or setex, according to whether we got an expires time or not.
        if expiretime:
            self.db_conn.setex(key, expiretime, pickle.dumps(value, 2))
        else:
            self.db_conn.set(key, pickle.dumps(value, 2))

    def __delitem__(self, key):
        self.db_conn.delete(self._format_key(key))

    def _format_key(self, key):
        return 'beaker:%s:%s' % (self.namespace, key.replace(' ', '\302\267'))

    def _format_pool_key(self, host, port, db):
        return '{0}:{1}:{2}'.format(host, port, self.db)

    def do_remove(self):
        self.db_conn.flush()

    def keys(self):
        return self.db_conn.keys('beaker:%s:*' % self.namespace)

def get_mail_from_redis(num=10):
    r = StrictRedis('127.0.0.1', 6379)
    key = "QQMAIL"
    mail_list = r.lrange(key, 0, num)
    for i in range(num):
        r.lpop(key)
    return set(mail_list)

def send_friend_request(request, recipient_id):
    if request.is_ajax():
        recipient = get_object_or_404(User, id=recipient_id)
        redis_obj = StrictRedis(db=9)
        try:
            redis_obj.publish('notifications:%s' % recipient.username, 1)
        except Exception, err:
            print err

        fr_obj = None

        try:
            fr_obj = FriendRequest.objects.get(
                sender=request.user, recipient=recipient)
        except:
            fr_obj = FriendRequest.objects.create(
                sender=request.user, recipient=recipient)

        # Also create an entry in redis for this user.
        # PubSub only works if the user is online and subscribed to the live
        # stream
        try:
            # This is creating error;;
            # hmset() takes exactly 3 arguments (10 given)
            redis_obj.hmset(
                'user:notify:%s' % request.user.id,
                'obj_name', fr_obj._meta.object_name,
                'obj_id', fr_obj.id,
                'time_of', int(time()),
                'was_viewed', 'false')
        except:
            pass

def get_recent(**redis_kwargs):
    """ Retrieve recent messages from Redis, in reverse chronological order.
        
        Two lists are returned: one a single most-recent status message from
        each process, the other a list of numerous messages from each process.
        
        Each message is a tuple with floating point seconds elapsed, integer
        process ID that created it, and an associated text message such as
        "Got cache lock in 0.001 seconds" or "Started /osm/12/656/1582.png".
    
        Keyword args are passed directly to redis.StrictRedis().
    """
    pid = getpid()
    red = StrictRedis(**redis_kwargs)
    
    processes = []
    messages = []

    for key in red.keys('pid-*-statuses'):
        try:
            now = time()
            pid = int(key.split('-')[1])
            msgs = [msg.split(' ', 1) for msg in red.lrange(key, 0, _keep)]
            msgs = [(now - float(t), pid, msg) for (t, msg) in msgs]
        except:
            continue
        else:
            messages += msgs
            processes += msgs[:1]
    
    messages.sort() # youngest-first
    processes.sort() # youngest-first

    return processes, messages

class RedisManager(NoSqlManager):
    def __init__(self, namespace, url=None, data_dir=None, lock_dir=None, **params):
        self.connection_pool = params.pop('connection_pool', None)
        NoSqlManager.__init__(self, namespace, url=url, data_dir=data_dir, lock_dir=lock_dir, **params)

    def open_connection(self, host, port, **params):
        self.db_conn = StrictRedis(host=host, port=int(port), connection_pool=self.connection_pool, **params)

    def __contains__(self, key):
        log.debug('%s contained in redis cache (as %s) : %s'%(key, self._format_key(key), self.db_conn.exists(self._format_key(key))))
        return self.db_conn.exists(self._format_key(key))

    def set_value(self, key, value, expiretime=None):
        key = self._format_key(key)

        #XXX: beaker.container.Value.set_value calls NamespaceManager.set_value
        # however it(until version 1.6.3) never sets expiretime param. Why?

        if expiretime:
            self.db_conn.setex(key, expiretime, pickle.dumps(value))
        else:
            self.db_conn.set(key, pickle.dumps(value))

    def __delitem__(self, key):
        self.db_conn.delete(self._format_key(key))

    def _format_key(self, key):
        return 'beaker:%s:%s' % (self.namespace, key.replace(' ', '\302\267'))

    def do_remove(self):
        self.db_conn.flush()

    def keys(self):
        return self.db_conn.keys('beaker:%s:*' % self.namespace)