[PLUGINS] Added DBQueue plugin
This commit is contained in:
parent
e3c34c4693
commit
333b915740
|
@ -59,7 +59,6 @@ abstract class QueueManager extends IoManager
|
||||||
if (empty(self::$qm)) {
|
if (empty(self::$qm)) {
|
||||||
|
|
||||||
if (Event::handle('StartNewQueueManager', array(&self::$qm))) {
|
if (Event::handle('StartNewQueueManager', array(&self::$qm))) {
|
||||||
|
|
||||||
$enabled = common_config('queue', 'enabled');
|
$enabled = common_config('queue', 'enabled');
|
||||||
$type = common_config('queue', 'subsystem');
|
$type = common_config('queue', 'subsystem');
|
||||||
|
|
||||||
|
@ -68,9 +67,6 @@ abstract class QueueManager extends IoManager
|
||||||
self::$qm = new UnQueueManager();
|
self::$qm = new UnQueueManager();
|
||||||
} else {
|
} else {
|
||||||
switch ($type) {
|
switch ($type) {
|
||||||
case 'db':
|
|
||||||
self::$qm = new DBQueueManager();
|
|
||||||
break;
|
|
||||||
default:
|
default:
|
||||||
throw new ServerException("No queue manager class for type '$type'");
|
throw new ServerException("No queue manager class for type '$type'");
|
||||||
}
|
}
|
||||||
|
|
49
plugins/DBQueue/DBQueuePlugin.php
Normal file
49
plugins/DBQueue/DBQueuePlugin.php
Normal file
|
@ -0,0 +1,49 @@
|
||||||
|
<?php
|
||||||
|
// This file is part of GNU social - https://www.gnu.org/software/social
|
||||||
|
//
|
||||||
|
// GNU social is free software: you can redistribute it and/or modify
|
||||||
|
// it under the terms of the GNU Affero General Public License as published by
|
||||||
|
// the Free Software Foundation, either version 3 of the License, or
|
||||||
|
// (at your option) any later version.
|
||||||
|
//
|
||||||
|
// GNU social is distributed in the hope that it will be useful,
|
||||||
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
// GNU Affero General Public License for more details.
|
||||||
|
//
|
||||||
|
// You should have received a copy of the GNU Affero General Public License
|
||||||
|
// along with GNU social. If not, see <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
/**
|
||||||
|
* DB interface for GNU social queues
|
||||||
|
*
|
||||||
|
* @package GNUsocial
|
||||||
|
* @author Miguel Dantas <biodantasgs@gmail.com>
|
||||||
|
* @copyright 2019 Free Software Foundation, Inc http://www.fsf.org
|
||||||
|
* @license https://www.gnu.org/licenses/agpl.html GNU AGPL v3 or later
|
||||||
|
*/
|
||||||
|
|
||||||
|
defined('GNUSOCIAL') || die();
|
||||||
|
|
||||||
|
class DBQueuePlugin extends Plugin
|
||||||
|
{
|
||||||
|
const PLUGIN_VERSION = '0.0.1';
|
||||||
|
|
||||||
|
public function onStartNewQueueManager(?QueueManager &$qm)
|
||||||
|
{
|
||||||
|
common_debug("Starting DB queue manager.");
|
||||||
|
$qm = new DBQueueManager();
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
public function onPluginVersion(array &$versions): bool
|
||||||
|
{
|
||||||
|
$versions[] = array('name' => 'DBQueue',
|
||||||
|
'version' => self::PLUGIN_VERSION,
|
||||||
|
'author' => 'Miguel Dantas',
|
||||||
|
'description' =>
|
||||||
|
// TRANS: Plugin description.
|
||||||
|
_m('Plugin using the database as a backend for GNU social queues'));
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
};
|
18
plugins/DBQueue/README
Normal file
18
plugins/DBQueue/README
Normal file
|
@ -0,0 +1,18 @@
|
||||||
|
DBQueuePlugin wraps the DBQueueManager class which is a queue manager
|
||||||
|
that uses the database as it's backing storage.
|
||||||
|
|
||||||
|
Installation
|
||||||
|
============
|
||||||
|
|
||||||
|
This plugin is replaces other queue manager plugins, such as UnQueue,
|
||||||
|
which enabled by default and which should, but is not required to be
|
||||||
|
disabled.
|
||||||
|
|
||||||
|
addPlugin('DBQueue');
|
||||||
|
|
||||||
|
Example
|
||||||
|
=======
|
||||||
|
|
||||||
|
In config.php
|
||||||
|
|
||||||
|
addPlugin('DBQueue');
|
|
@ -50,13 +50,13 @@ class DBQueueManager extends QueueManager
|
||||||
}
|
}
|
||||||
|
|
||||||
$this->stats('enqueued', $queue);
|
$this->stats('enqueued', $queue);
|
||||||
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Poll every 10 seconds for new events during idle periods.
|
* Poll every 10 seconds for new events during idle periods.
|
||||||
* We'll look in more often when there's data available.
|
* We'll look in more often when there's data available.
|
||||||
|
* Must be greater than 0 for the poll method to be called
|
||||||
*
|
*
|
||||||
* @return int seconds
|
* @return int seconds
|
||||||
*/
|
*/
|
||||||
|
@ -69,9 +69,9 @@ class DBQueueManager extends QueueManager
|
||||||
* Run a polling cycle during idle processing in the input loop.
|
* Run a polling cycle during idle processing in the input loop.
|
||||||
* @return boolean true if we should poll again for more data immediately
|
* @return boolean true if we should poll again for more data immediately
|
||||||
*/
|
*/
|
||||||
public function poll()
|
public function poll(): bool
|
||||||
{
|
{
|
||||||
//$this->_log(LOG_DEBUG, 'Checking for notices...');
|
$this->_log(LOG_DEBUG, 'Checking for notices...');
|
||||||
$qi = Queue_item::top($this->activeQueues(), $this->getIgnoredTransports());
|
$qi = Queue_item::top($this->activeQueues(), $this->getIgnoredTransports());
|
||||||
if (!$qi instanceof Queue_item) {
|
if (!$qi instanceof Queue_item) {
|
||||||
//$this->_log(LOG_DEBUG, 'No notices waiting; idling.');
|
//$this->_log(LOG_DEBUG, 'No notices waiting; idling.');
|
||||||
|
@ -96,13 +96,16 @@ class DBQueueManager extends QueueManager
|
||||||
$this->noHandlerFound($qi, $rep);
|
$this->noHandlerFound($qi, $rep);
|
||||||
return true;
|
return true;
|
||||||
} catch (NoResultException $e) {
|
} catch (NoResultException $e) {
|
||||||
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('._ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
|
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('.
|
||||||
|
_ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
|
||||||
$result = true;
|
$result = true;
|
||||||
} catch (AlreadyFulfilledException $e) {
|
} catch (AlreadyFulfilledException $e) {
|
||||||
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('._ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
|
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] ".get_class($e).' thrown ('.
|
||||||
|
_ve($e->getMessage()).'), ignoring queue_item '._ve($qi->getID()));
|
||||||
$result = true;
|
$result = true;
|
||||||
} catch (Exception $e) {
|
} catch (Exception $e) {
|
||||||
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] Exception (".get_class($e).') thrown: '._ve($e->getMessage()));
|
$this->_log(LOG_ERR, "[{$qi->transport}:$rep] Exception (".
|
||||||
|
get_class($e).') thrown: '._ve($e->getMessage()));
|
||||||
$result = false;
|
$result = false;
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,6 +10,7 @@ class RedisQueueManager extends QueueManager
|
||||||
|
|
||||||
public function __construct(string $server)
|
public function __construct(string $server)
|
||||||
{
|
{
|
||||||
|
parent::__construct();
|
||||||
$this->server = $server;
|
$this->server = $server;
|
||||||
$this->queue = 'gnusocial:' . common_config('site', 'name');
|
$this->queue = 'gnusocial:' . common_config('site', 'name');
|
||||||
}
|
}
|
||||||
|
@ -39,7 +40,6 @@ class RedisQueueManager extends QueueManager
|
||||||
|
|
||||||
public function poll()
|
public function poll()
|
||||||
{
|
{
|
||||||
common_debug("STARTING POLL");
|
|
||||||
try {
|
try {
|
||||||
$this->_ensureConn();
|
$this->_ensureConn();
|
||||||
$ret = $this->client->lpop($this->queue);
|
$ret = $this->client->lpop($this->queue);
|
||||||
|
|
Loading…
Reference in New Issue
Block a user