JobQueueAggregatorRedis.php

Go to the documentation of this file.

<?php
class JobQueueAggregatorRedis extends JobQueueAggregator {
        protected $redisPool;

        protected function __construct( array $params ) {
                parent::__construct( $params );
                $this->server = $params['redisServer'];
                $this->redisPool = RedisConnectionPool::singleton( $params['redisConfig'] );
        }

        protected function doNotifyQueueEmpty( $wiki, $type ) {
                $conn = $this->getConnection();
                if ( !$conn ) {
                        return false;
                }
                try {
                        $conn->hDel( $this->getReadyQueueKey(), $this->encQueueName( $type, $wiki ) );
                        return true;
                } catch ( RedisException $e ) {
                        $this->handleException( $conn, $e );
                        return false;
                }
        }

        protected function doNotifyQueueNonEmpty( $wiki, $type ) {
                $conn = $this->getConnection();
                if ( !$conn ) {
                        return false;
                }
                try {
                        $conn->hSet( $this->getReadyQueueKey(), $this->encQueueName( $type, $wiki ), time() );
                        return true;
                } catch ( RedisException $e ) {
                        $this->handleException( $conn, $e );
                        return false;
                }
        }

        protected function doGetAllReadyWikiQueues() {
                $conn = $this->getConnection();
                if ( !$conn ) {
                        return array();
                }
                try {
                        $conn->multi( Redis::PIPELINE );
                        $conn->exists( $this->getReadyQueueKey() );
                        $conn->hGetAll( $this->getReadyQueueKey() );
                        list( $exists, $map ) = $conn->exec();

                        if ( $exists ) { // cache hit
                                $pendingDBs = array(); // (type => list of wikis)
                                foreach ( $map as $key => $time ) {
                                        list( $type, $wiki ) = $this->dencQueueName( $key );
                                        $pendingDBs[$type][] = $wiki;
                                }
                        } else { // cache miss
                                $pendingDBs = $this->findPendingWikiQueues(); // (type => list of wikis)

                                $now = time();
                                $map = array();
                                foreach ( $pendingDBs as $type => $wikis ) {
                                        foreach ( $wikis as $wiki ) {
                                                $map[$this->encQueueName( $type, $wiki )] = $now;
                                        }
                                }
                                $conn->hMSet( $this->getReadyQueueKey(), $map );
                        }

                        return $pendingDBs;
                } catch ( RedisException $e ) {
                        $this->handleException( $conn, $e );
                        return array();
                }
        }

        protected function getConnection() {
                return $this->redisPool->getConnection( $this->server );
        }

        protected function handleException( RedisConnRef $conn, $e ) {
                $this->redisPool->handleException( $this->server, $conn, $e );
        }

        private function getReadyQueueKey() {
                return "jobqueue:aggregator:h-ready-queues:v1"; // global
        }

        private function encQueueName( $type, $wiki ) {
                return rawurlencode( $type ) . '/' . rawurlencode( $wiki );
        }

        private function dencQueueName( $name ) {
                list( $type, $wiki ) = explode( '/', $name, 2 );
                return array( rawurldecode( $type ), rawurldecode( $wiki ) );
        }
}