Merge branch 'dev' of https://framagit.org/hubzilla/core into xdev_merge

2 files changed, 23 insertions, 121 deletions

diff --git a/Zotlabs/Daemon/Master.php b/Zotlabs/Daemon/Master.php
index 0656ca05b..857d47243 100644
--- a/Zotlabs/Daemon/Master.php
+++ b/Zotlabs/Daemon/Master.php
@@ -16,8 +16,6 @@ if(array_search( __file__ , get_included_files()) === 0) {
 
 class Master {
 
-	static public $queueworker = null;
-
 	static public function Summon($arr) {
 		proc_run('php','Zotlabs/Daemon/Master.php',$arr);
 	}
@@ -25,126 +23,21 @@ class Master {
 	static public function Release($argc,$argv) {
 		cli_startup();
 
-		$maxworkers = get_config('system','max_queue_workers');
-
-		if (!$maxworkers || $maxworkers == 0) {
-			logger('Master: release: ' . print_r($argv,true), LOGGER_ALL,LOG_DEBUG);
-                	$cls = '\\Zotlabs\\Daemon\\' . $argv[0];
-                	$cls::run($argc,$argv);
-			self::ClearQueue();
-		} else {
-			logger('Master: enqueue: ' . print_r($argv,true), LOGGER_ALL,LOG_DEBUG);
-			$workinfo = ['argc'=>$argc,'argv'=>$argv];
-			q("insert into config (cat,k,v) values ('queuework','%s','%s')",
-				dbesc(uniqid('workitem:',true)),
-				dbesc(serialize($workinfo)));
-			self::Process();
-		}
-	}
-
-	static public function GetWorkerID() {
-		$maxworkers = get_config('system','max_queue_workers');
-		$maxworkers = ($maxworkers) ? $maxworkers : 3;
-
-		$workermaxage = get_config('system','max_queue_worker_age');
-		$workermaxage = ($workermaxage) ? $workermaxage : 300;
-
-		$workers = q("select * from config where cat='queueworkers' and k like '%s'", 'workerstarted_%');
-
-		if (count($workers) > $maxworkers) {
-			foreach ($workers as $idx => $worker) {
-                                $curtime = time();
-                                $age = (intval($curtime) - intval($worker['v']));
-                                if ( $age > $workermaxage) {
-					logger("Prune worker: ".$worker['k'], LOGGER_ALL, LOGGER_DEBUG);
-					$k = explode('_',$worker['k']);
-					q("delete from config where cat='queueworkers' and k='%s'",
-						'workerstarted_'.$k[1]);
-					q("update config set k='%s' where cat='queuework' and k='%s'",
-						dbesc(uniqid('workitem:',true)),
-						'workitem_'.$k[1]);
-					unset($workers[$idx]);
-				}
-			}
-			if (count($workers) > $maxworkers) {
-				return false;
-			}
-		}
-		return uniqid('',true);
-
-	}
-
-	static public function Process() {
-
-		self::$queueworker = self::GetWorkerID();
-
-		if (!self::$queueworker) {
-			logger('Master: unable to obtain worker ID.');
-			killme();
-		}
-
-		set_config('queueworkers','workerstarted_'.self::$queueworker,time());
-
-		$workersleep = get_config('system','queue_worker_sleep');
-		$workersleep = ($workersleep) ? $workersleep : 5;
-		cli_startup();
-
-		$work = q("update config set k='%s' where cat='queuework' and k like '%s' limit 1",
-			'workitem_'.self::$queueworker,
-			dbesc('workitem:%'));
-		$jobs = 0;
-		while ($work) {
-			$workitem = q("select * from config where cat='queuework' and k='%s'",
-				'workitem_'.self::$queueworker);
-
-			if (isset($workitem[0])) {
-				$jobs++;
-				$workinfo = unserialize($workitem[0]['v']);
-				$argc = $workinfo['argc'];
-				$argv = $workinfo['argv'];
-				logger('Master: process: ' . print_r($argv,true), LOGGER_ALL,LOG_DEBUG);
-
-				//Delete unclaimed duplicate workitems.
-				q("delete from config where cat='queuework' and k='workitem' and v='%s'",
-					serialize($argv));
-
-				$cls = '\\Zotlabs\\Daemon\\' . $argv[0];
-				$cls::run($argc,$argv);
+		$hookinfo = [
+			'argv'=>$argv
+		];
 
-				//Right now we assume that if we get a return, everything is OK.
-				//At some point we may want to test whether the run returns true/false
-				//    and requeue the work to be tried again.  But we probably want
-				//    to implement some sort of "retry interval" first.
+		call_hooks ('daemon_master_release',$hookinfo);
 
-				q("delete from config where cat='queuework' and k='%s'",
-					'workitem_'.self::$queueworker);
-			} else {
-				break;
-			}
-			sleep ($workersleep);
-			$work = q("update config set k='%s' where cat='queuework' and k like '%s' limit 1",
-				'workitem_'.self::$queueworker,
-				dbesc('workitem:%'));
+		$argv = $hookinfo['argv'];
+		$argc = count($argv);
 
+		if ((!is_array($argv) || (count($argv) < 1))) {
+			return;
 		}
-		logger('Master: Worker Thread: queue items processed:' . $jobs);
-		q("delete from config where cat='queueworkers' and k='%s'",
-			'workerstarted_'.self::$queueworker);
-	}	
 
-	static public function ClearQueue() {
-			$work = q("select * from config where cat='queuework' and k like '%s'",
-				dbesc('workitem%'));
-			foreach ($work as $workitem) {
-				$workinfo = unserialize($workitem['v']);
-				$argc = $workinfo['argc'];
-				$argv = $workinfo['argv'];
-				logger('Master: process: ' . print_r($argv,true), LOGGER_ALL,LOG_DEBUG);
-				$cls = '\\Zotlabs\\Daemon\\' . $argv[0];
-				$cls::run($argc,$argv);
-			}
-			$work = q("delete from config where cat='queuework' and k like '%s'",
-				dbesc('workitem%'));
+		logger('Master: release: ' . json_encode($argv), LOGGER_ALL,LOG_DEBUG);
+                $cls = '\\Zotlabs\\Daemon\\' . $argv[0];
+                $cls::run($argc,$argv);
 	}
-	
 }
diff --git a/Zotlabs/Daemon/Queue.php b/Zotlabs/Daemon/Queue.php
index 8f529ff13..814148404 100644
--- a/Zotlabs/Daemon/Queue.php
+++ b/Zotlabs/Daemon/Queue.php
@@ -12,7 +12,6 @@ class Queue {
 		require_once('include/items.php');
 		require_once('include/bbcode.php');
 
-
 		if($argc > 1)
 			$queue_id = $argv[1];
 		else
@@ -61,10 +60,20 @@ class Queue {
 			// or just prior to this query based on recent and long-term delivery history. If we have good reason to believe
 			// the site is permanently down, there's no reason to attempt delivery at all, or at most not more than once 
 			// or twice a day. 
-	
-			$r = q("SELECT * FROM outq WHERE outq_delivered = 0 and outq_scheduled < %s ",
+
+			$sqlrandfunc = db_getfunc('rand');
+
+			$r = q("SELECT *,$sqlrandfunc as rn FROM outq WHERE outq_delivered = 0 and outq_scheduled < %s order by rn limit 1",
 				db_utcnow()
 			);
+			while ($r) {
+				foreach($r as $rv) {
+					queue_deliver($rv);
+				}
+				$r = q("SELECT *,$sqlrandfunc as rn FROM outq WHERE outq_delivered = 0 and outq_scheduled < %s order by rn limit 1",
+					db_utcnow()
+				);
+			}
 		}
 		if(! $r)
 			return;