synchronizes messages after redis reconnect between server Instances
172309
function synchronize() {
    if (syncPointer != -1) {
        log.logMessage("verbose", "redis synchronization!");
        var syncMessage = {};
        syncMessage.data = messages.splice(syncPointer, messages.length - syncPointer);
        setTimeout(function () {
            pub.publish("message", JSON.stringify(syncMessage));
        }, 200);
        syncPointer = -1;
    }
}
876264
def setnx(msg_id, suffix, value)
      with_failover { redis.setnx(key(msg_id, suffix), value) }
    end
1248886
public static function redis()
    {
        // Detect when the PID of the current process has changed (from a fork, etc)
        // and force a reconnect to redis.
        $pid = getmypid();
        if (self::$pid !== $pid) {
            self::$redis = null;
            self::$pid = $pid;
        }

        if (!is_null(self::$redis)) {
            return self::$redis;
        }

        $server = self::$redisServer;
        if (empty($server)) {
            $server = 'localhost:6379';
        }

        if (is_array($server)) {
            require_once dirname(__FILE__) . '/Resque/RedisCluster.php';
            self::$redis = new Resque_RedisCluster($server);
        } else {
            if (strpos($server, 'unix:') === false) {
                list($host, $port) = explode(':', $server);
            } else {
                $host = $server;
                $port = null;
            }
            require_once dirname(__FILE__) . '/Resque/Redis.php';
            $redisInstance = new Resque_Redis($host, $port, self::$password);
            $redisInstance->prefix(self::$namespace);
            self::$redis = $redisInstance;
        }

        if (!empty(self::$redisDatabase)) {
            self::$redis->select(self::$redisDatabase);
        }

        return self::$redis;
    }
512513
void reconnect() throws InterruptedException {
        long maxTries = options.getMaxReconnect();
        long tries = 0;
        String lastServer = null;

        if (isClosed()) {
            return;
        }

        if (maxTries == 0) {
            this.close();
            return;
        }

        this.writer.setReconnectMode(true);

        while (!isConnected() && !isClosed() && !this.isClosing()) {
            Collection<String> serversToTry = buildReconnectList();

            for (String server : serversToTry) {
                if (isClosed()) {
                    break;
                }

                if (server.equals(lastServer)) {
                    this.reconnectWaiter = new CompletableFuture<>();
                    waitForReconnectTimeout();
                }

                if (isDisconnectingOrClosed() || this.isClosing()) {
                    break;
                }

                updateStatus(Status.RECONNECTING);

                tryToConnect(server);
                lastServer = server;
                tries++;

                if (maxTries > 0 && tries >= maxTries) {
                    break;
                } else if (isConnected()) {
                    this.statistics.incrementReconnects();
                    break;
                }
            }

            if (maxTries > 0 && tries >= maxTries) {
                break;
            }
        }

        if (!isConnected()) {
            this.close();
            return;
        }

        this.subscribers.forEach((sid, sub) -> {
            if (sub.getDispatcher() == null && !sub.isDraining()) {
                sendSubscriptionMessage(sub.getSID(), sub.getSubject(), sub.getQueueName(), true);
            }
        });

        this.dispatchers.forEach((nuid, d) -> {
            if (!d.isDraining()) {
                d.resendSubscriptions();
            }
        });

        try {
            this.flush(this.options.getConnectionTimeout());
        } catch (Exception exp) {
            this.processException(exp);
        }
        
        // When the flush returns we are done sending internal messages, so we can switch to the
        // non-reconnect queue
        this.writer.setReconnectMode(false);

        processConnectionEvent(Events.RESUBSCRIBED);
    }
1328226
def work_wait(self):
        

        if len(self.queues_with_notify) > 0:
            # https://github.com/antirez/redis/issues/874
            connections.redis.blpop(*(self.queues_with_notify + [max(1, int(self.config["max_latency"]))]))
        else:
            gevent.sleep(self.config["max_latency"])
606820
def listen_for_events():
    
    import_event_modules()
    conn = redis_connection.get_connection()
    pubsub = conn.pubsub()
    pubsub.subscribe("eventlib")
    for message in pubsub.listen():
        if message['type'] != 'message':
            continue
        data = loads(message["data"])
        if 'name' in data:
            event_name = data.pop('name')
            process_external(event_name, data)
426655
def observing_connection(&block)
      yield

    rescue *connection_errors => exception
      if defined?(Trouble)
        if Trouble.respond_to? :log  # Requires Trouble > 0.0.11
          Trouble.log exception, code: :lost_connection, message: 'Raidis lost connection to the Redis server.', client: redis.inspect
        else
          Trouble.notify exception, code: :lost_connection, message: 'Raidis lost connection to the Redis server.', client: redis.inspect
        end
      end

      raise Raidis::ConnectionError, exception

    rescue Redis::CommandError => exception
      if exception.message.to_s.split.first == 'READONLY'
        Trouble.notify(exception, code: :readonly, message: 'Raidis detected an illegal write against a Redis slave.', client: redis.inspect) if defined?(Trouble)
        raise Raidis::ConnectionError, exception
      else
        # Passing through Exceptions unrelated to the Connection. E.g. Redis::CommandError.
        raise exception
      end
    end
173047
public boolean redeliverLocked( TransactionItem[] items , MessageLockSet locks ) throws JMSException
    {
    	checkNotClosed();
    	checkTransactionLock();
    	
    	int volatileRollbacked = 0;
    	int persistentRollbacked = 0;

        synchronized (storeLock)
        {
            for (int n = 0 ; n < items.length ; n++)
            {
            	TransactionItem transactionItem = items[n];
            	if (transactionItem.getDestination() != this)
            		continue; // Not for us
            	
            	MessageStore store = transactionItem.getDeliveryMode() == DeliveryMode.PERSISTENT ? persistentStore : volatileStore;
            	int handle = transactionItem.getHandle();
            	
            	// Retrieve message content
           		AbstractMessage msg = store.retrieve(handle);
           		
           		// Update redelivered flag both in memory and message store
           		msg.setJMSRedelivered(true);
           		handle = store.replace(handle, msg);
           		
            	if (redeliveryDelay > 0)
            	{
            	    // Keep the message locked so it cannot be re-consumed immediately            	    
            		// and schedule message unlock after redeliveryDelay milliseconds
            		redeliveryTimer.schedule(new RedeliveryTask(msg,store,handle), 
            				                 redeliveryDelay);
            	}
            	else
            	{
            		// Store lock for later release
            		locks.add(handle, store.getDeliveryMode(), this, msg);
            	}

                if (transactionItem.getDeliveryMode() == DeliveryMode.PERSISTENT)
                	persistentRollbacked++;
                else
                	volatileRollbacked++;
            }
        }
        rollbackedGetCount.addAndGet(volatileRollbacked + persistentRollbacked);

        if (persistentRollbacked > 0 && requiresTransactionalUpdate())
        {
        	pendingChanges = true;
        	return true;
        }
        else
        	return false;
    }
1356206
private AsyncConnections getConnections(Iterable<RedisNodeDescription> nodes) {

        List<RedisNodeDescription> nodeList = LettuceLists.newList(nodes);
        AsyncConnections connections = new AsyncConnections(nodeList);

        for (RedisNodeDescription node : nodeList) {

            RedisURI redisURI = node.getUri();
            String message = String.format("Unable to connect to %s", redisURI);
            try {
                CompletableFuture<StatefulRedisConnection<String, String>> connectionFuture = nodeConnectionFactory
                        .connectToNodeAsync(CODEC, redisURI);

                CompletableFuture<StatefulRedisConnection<String, String>> sync = new CompletableFuture<>();

                connectionFuture.whenComplete((connection, throwable) -> {

                    if (throwable != null) {

                        if (throwable instanceof RedisConnectionException) {
                            if (logger.isDebugEnabled()) {
                                logger.debug(throwable.getMessage(), throwable);
                            } else {
                                logger.warn(throwable.getMessage());
                            }
                        } else {
                            logger.warn(message, throwable);
                        }

                        sync.completeExceptionally(new RedisConnectionException(message, throwable));
                    } else {
                        connection.async().clientSetname("lettuce#MasterSlaveTopologyRefresh");
                        sync.complete(connection);
                    }
                });

                connections.addConnection(redisURI, sync);
            } catch (RuntimeException e) {
                logger.warn(String.format(message, redisURI), e);
            }
        }

        return connections;
    }
1131008
def redis_client(address=None, connection_pool=None, timeout=None,
                 parser=None, **kwargs):
    '''
    '''
    if not connection_pool:
        if timeout == 0:
            if not async:
                raise ImportError('Asynchronous connection requires async '
                                  'bindings installed.')
            return async.pool.redis(address, **kwargs)
        else:
            kwargs['socket_timeout'] = timeout
            return Redis(address[0], address[1], **kwargs)
    else:
        return Redis(connection_pool=connection_pool)
9242
def mset(msg_id, values)
      values = values.inject([]){|a,(k,v)| a.concat([key(msg_id, k), v])}
      with_failover { redis.mset(*values) }
    end
476210
def _check_and_send(self):
        
        if self.transport._stop_event.ready() or not self.transport.greenlet:
            self.log.error("Can't retry - stopped")
            return

        if self.transport._prioritize_global_messages:
            # During startup global messages have to be sent first
            self.transport._global_send_queue.join()

        self.log.debug('Retrying message', receiver=to_normalized_address(self.receiver))
        status = self.transport._address_mgr.get_address_reachability(self.receiver)
        if status is not AddressReachability.REACHABLE:
            # if partner is not reachable, return
            self.log.debug(
                'Partner not reachable. Skipping.',
                partner=pex(self.receiver),
                status=status,
            )
            return
        # sort output by channel_identifier (so global/unordered queue goes first)
        # inside queue, preserve order in which messages were enqueued
        ordered_queue = sorted(
            self._message_queue,
            key=lambda d: d.queue_identifier.channel_identifier,
        )
        message_texts = [
            data.text
            for data in ordered_queue
            # if expired_gen generator yields False, message was sent recently, so skip it
            if next(data.expiration_generator)
        ]

        def message_is_in_queue(data: _RetryQueue._MessageData) -> bool:
            return any(
                isinstance(data.message, RetrieableMessage) and
                send_event.message_identifier == data.message.message_identifier
                for send_event in self.transport._queueids_to_queues[data.queue_identifier]
            )

        # clean after composing, so any queued messages (e.g. Delivered) are sent at least once
        for msg_data in self._message_queue[:]:
            remove = False
            if isinstance(msg_data.message, (Delivered, Ping, Pong)):
                # e.g. Delivered, send only once and then clear
                # TODO: Is this correct? Will a missed Delivered be 'fixed' by the
                #       later `Processed` message?
                remove = True
            elif msg_data.queue_identifier not in self.transport._queueids_to_queues:
                remove = True
                self.log.debug(
                    'Stopping message send retry',
                    queue=msg_data.queue_identifier,
                    message=msg_data.message,
                    reason='Raiden queue is gone',
                )
            elif not message_is_in_queue(msg_data):
                remove = True
                self.log.debug(
                    'Stopping message send retry',
                    queue=msg_data.queue_identifier,
                    message=msg_data.message,
                    reason='Message was removed from queue',
                )

            if remove:
                self._message_queue.remove(msg_data)

        if message_texts:
            self.log.debug('Send', receiver=pex(self.receiver), messages=message_texts)
            self.transport._send_raw(self.receiver, '\n'.join(message_texts))
345568
async def redis(self) -> aioredis.Redis:
        
        # Use thread-safe asyncio Lock because this method without that is not safe
        async with self._connection_lock:
            if self._redis is None:
                self._redis = await aioredis.create_redis_pool((self._host, self._port),
                                                               db=self._db, password=self._password, ssl=self._ssl,
                                                               minsize=1, maxsize=self._pool_size,
                                                               loop=self._loop, **self._kwargs)
        return self._redis
553436
def poll(self):
        

        pubsub = self.conn.pubsub()
        pubsub.subscribe(self.key)

        # Pubsub messages generator
        gen = pubsub.listen()

        last_heartbeat = time.time()
        self.heartbeat()

        last_cleanup = time.time()
        self.cleanup()

        try:
            while True:
                # Since Redis' listen method blocks, we use select to inspect the
                # underlying socket to see if there is activity.
                fileno = pubsub.connection._sock.fileno()
                timeout = max(0, POLL_INTERVAL - (time.time() - last_heartbeat))
                r, w, x = self._select([fileno], [], [], timeout)
                if fileno in r:
                    next(gen)
                    self.update()

                last_heartbeat = time.time()
                self.heartbeat()

                now = time.time()
                if now - last_cleanup > CLEANUP_INTERVAL:
                    last_cleanup = now
                    self.cleanup()
        finally:
            pubsub.close()
458915
public synchronized void sendRetry(long retry, EventCallback callback) {

        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, null, null, null, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(retry, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }
1200266
public void sync() throws Exception {
        try {
            JedisClusterPipeline pipeline = new JedisClusterPipeline(clusterInfoCache);
            execute(pipeline);
            pipeline.sync();
        } catch (JedisMovedDataException jre) {
            // if MOVED redirection occurred, rebuilds cluster's slot cache,
            // recommended by Redis cluster specification
            connectionHandler.renewSlotCache();
            if (maxAttempts > 0) {
                maxAttempts--;
                sync();
                return;
            }
            throw jre;
        } catch (Exception e) {
            throw e;
        }
    }
564921
def after_init_apps(sender):
    
    from uliweb import settings
    from uliweb.utils.common import log
    
    check = settings.get_var('REDIS/check_version')
    if check:
        client = get_redis()
        try:
            info = client.info()
        except Exception as e:
            log.exception(e)
            log.error('Redis is not started!')
            return
        
        redis_version = info['redis_version']
        version = tuple(map(int, redis_version.split('.')))
        op = re_compare_op.search(check)
        if op:
            _op = op.group()
            _v = check[op.end()+1:].strip()
        else:
            _op = '='
            _v = check
        nv = tuple(map(int, _v.split('.')))
        if _op == '=':
            flag = version[:len(nv)] == nv
        elif _op == '>=':
            flag = version >= nv
        elif _op == '>':
            flag = version > nv
        elif _op == '<=':
            flag = version <= nv
        elif _op == '<':
            flag = version < nv
        else:
            log.error("Can't support operator %s when check redis version" % _op)
        if not flag:
            log.error("Redis version %s is not matched what you want %s" % (redis_version, _v))