/inc/events.inc.php (6859bd8edb85707a8aa933aca9ca0cc5a48e4c42) (6321 bytes) (mode 100644) (type blob)

<?php
//
// This functions are used for background tasks: the tasks that the user should
// not wait to happen in the browser: e-mails, keyring regeneration etc.
//
require_once($INC . "/util.inc.php");
require_once($INC . "/log.inc.php");
require_once($INC . "/sql.inc.php");
require_once($INC . "/prof.inc.php");

if (!isset($rg_event_socket))
	$rg_event_socket = "/var/lib/rocketgit/sockets/event.sock";

$rg_event_error = "";

function rg_event_set_error($str)
{
	global $rg_event_error;
	$rg_event_error = $str;
	rg_log($str);
}

function rg_event_error()
{
	global $rg_event_error;
	return $rg_event_error;
}

$rg_event_split_functions = array();

/*
 * Register event functions.
 */
function rg_event_register_functions($functions)
{
	global $rg_event_split_functions;

	if (empty($functions))
		return FALSE;

	foreach ($functions as $type => $function)
		$rg_event_split_functions[$type] = $function;

	return TRUE;
}

/*
 * Signals the daemon that there is some work to do
 * @timeout == NULL=forever, 0=no_wait, else, wait @timeout ms
 */
function rg_event_signal_daemon($ev_id, $timeout)
{
	global $rg_event_socket;

	if (empty($rg_event_socket)) {
		rg_log('DEBUG: rg_event_socket is not defined!');
		return TRUE;
	}

	if ($timeout === NULL)
		 $s_timeout = "forever";
	else if ($timeout === 0)
		$s_timeout = "no_wait";
	else
		$s_timeout = $timeout . "ms";

	rg_prof_start("event_signal_daemon");
	rg_log_enter("event_signal_daemon: event_id=[$ev_id] timeout=$s_timeout");

	if (empty($ev_id))
		$buf = "W\n";
	else
		$buf = "NOTIFY " . $ev_id . "\n";

	$flags = 0;
	if ($timeout === 0)
		$flags |= RG_SOCKET_NO_WAIT;

	$r = rg_socket($rg_event_socket, $buf, $timeout, 1, $flags);

	rg_log_exit();
	return $r;
}

/*
 * Inserts an event
 * This function is called from all over the place to generate events
 */
function rg_event_add($db, $event)
{
	global $rg_debug;

	rg_prof_start("event_add");
	rg_log_enter("event_add: event=" . rg_array2string($event));

	if (!isset($event['ip']))
		$event['ip'] = rg_var_str('REMOTE_ADDR');

	if (!isset($event['debug']))
		$event['debug'] = 0;
	if ($event['debug'] == 0)
		$event['debug'] = rg_var_uint('rg_debug');
	if ($rg_debug)
		$event['debug'] = 1;

	$ret = FALSE;
	while (1) {
		$now = time();
		$prio = $event['prio']; unset($event['prio']);
		$params = array("now" => $now,
			"prio" => $prio,
			"data" => serialize($event));
		$sql = "INSERT INTO events (itime, prio, data)"
			. " VALUES (@@now@@, @@prio@@, @@data@@)";
		$res = rg_sql_query_params($db, $sql, $params);
		if ($res === FALSE) {
			rg_event_set_error("Could not add event (" . rg_sql_error() . ")");
			break;
		}
		rg_sql_free_result($res);

		$ret = TRUE;
		break;
	}

	rg_log_exit();
	rg_prof_end("event_add");
	return $ret;
}

/*
 * Process an event
 */
function rg_event_process($db, $ev_id, $event)
{
	global $rg_event_split_functions;

	rg_prof_start("event_process");
	rg_log_enter("event_process: ev_id=$ev_id"
		. " event=" . rg_array2string($event));

	$ret = FALSE;
	while (1) {
		$category = $event['category'];
		unset($event['category']);

		if (!isset($rg_event_split_functions[$category])) {
			rg_event_set_error("Cannot find event function [cat=$category]!");
			rg_log_ml("DEBUG: rg_event_split_functions="
				. print_r($rg_event_split_functions, TRUE));
			break;
		}

		$f = $rg_event_split_functions[$category];
		rg_log("Calling $f...");
		$evs = $f($db, $event);
		if ($evs === FALSE) {
			rg_event_set_error('Error in function ['
				. $f . '] (category [' . $category . '])!');
			break;
		}

		if (!is_array($evs))
			rg_internal_error("evs is not array!");

		if (empty($evs)) {
			$ret = TRUE;
			break;
		}

		// TODO: here we must do a transaction?

		$r = TRUE;
		foreach ($evs as $index => $ev) {
			$r = rg_event_add($db, $ev);
			if ($r !== TRUE)
				break;
		}
		if ($r !== TRUE)
			break;

		$ret = TRUE;
		break;
	}

	rg_log_exit();
	rg_prof_end("event_process");
	return $ret;
}

/*
 * Notifies all listeners when @ev_id is happening.
 */
function rg_event_notify(&$notify_list, $ev_id, $misc)
{
	if (!isset($notify_list[$ev_id]))
		return;

	$buf = "DONE $ev_id $misc\n";
	$buf_len = strlen($buf);
	foreach ($notify_list[$ev_id] as $index => $info) {
		if (isset($info['func'])) {
			$info['func']($info['priv'], $buf);
			continue;
		}

		$fd = $info['fd'];
		rg_log("Notify [$ev_id] [fd=$fd]...");
		$r = @socket_send($fd, $buf, $buf_len, 0);
		if ($r === FALSE)
			rg_log("Error in sending.");
	}

	return;
}

/*
 * Process events queue
 * reset id to 1 if queue is empty?
 * Returns FALSE on error, else, the number of events processed
 * @notify_list: Will be used to signal the finish of an event
 */
function rg_event_process_queue($db, &$notify_list)
{
	rg_prof_start("event_process_queue");
	rg_log_enter("event_process_queue: notify_list: "
		. rg_array2string($notify_list));

	$ret = FALSE;
	while (1) {
		$now = time();

		// We limit to be able to deal with high prio tasks
		$sql = "SELECT * FROM events"
			. " WHERE fail = 0"
			. " AND next_try < $now"
			. " ORDER BY prio, id"
			. " FOR UPDATE"
			. " LIMIT 100";
		$res = rg_sql_query($db, $sql);
		if ($res === FALSE) {
			rg_event_set_error("Cannot load job list"
				. " (" . rg_sql_error() . ")");
			break;
		}

		$no_of_events = rg_sql_num_rows($res);
		while (($row = rg_sql_fetch_array($res))) {
			$params = array('id' => $row['id']);

			$sql = "UPDATE events SET fail = 1 WHERE id = @@id@@";
			while (1) {
				$ev = unserialize($row['data']);
				if ($ev === FALSE) {
					rg_internal_error("Cannot unserialize data");
					break;
				}

				$ev['prio'] = $row['prio'];
				$ev['itime'] = $row['itime'];

				$r = rg_event_process($db, $row['id'], $ev);
				if ($r !== TRUE) {
					if ($ev['debug'] == 1)
						break;

					$sql = "UPDATE events"
						. " SET tries = tries + 1"
						. ", next_try = $now + tries * 600"
						. " WHERE id = @@id@@";
					break;
				}

				if (isset($ev['notification']))
					rg_event_notify($notify_list, $ev['notification'], "");

				$sql = "DELETE FROM events WHERE id = @@id@@";
				break;
			}

			$res2 = rg_sql_query_params($db, $sql, $params);
			rg_sql_free_result($res2);
		}
		rg_sql_free_result($res);

		$ret = $no_of_events;
		break;
	}

	rg_log_exit();
	rg_prof_end("event_process_queue");
	return $ret;
}

?>


Mode Type Size Ref File
100644 blob 9 f3c7a7c5da68804a1bdf391127ba34aed33c3cca .exclude
100644 blob 102 eaeb7d777062c60a55cdd4b5734902cdf6e1790c .gitignore
100644 blob 289 fabbff669e768c05d6cfab4d9aeb651bf623e174 AUTHORS
100644 blob 1132 dd65951315f3de6d52d52a82fca59889d1d95187 Certs.txt
100644 blob 549 41c3bdbba8ec2523fe24b84bdd46777fc13e8345 History.txt
100644 blob 34520 dba13ed2ddf783ee8118c6a581dbf75305f816a3 LICENSE
100644 blob 2800 60d110e27bb04809efc8f5ccafc0fa543bf1e296 Makefile.in
100644 blob 4875 351369ca6f3895965cd98b847161c696d2052146 README
100644 blob 111473 53bf108ca16ba68ce66f0a02110a540c78e5d1ed TODO
100644 blob 1294 f22911eb777f0695fcf81ad686eac133eb11fcc4 TODO-plans
100644 blob 203 a2863c67c3da44126b61a15a6f09738c25e0fbe0 TODO.perf
100644 blob 600 5525d768c22262f90a504a11db4fabc25ddbab8f TODO.vm
040000 tree - 21928e906ad2907a55c2e81c2a8b0502b586b8a0 artwork
100644 blob 4207 693a5395a40ac2505dcfe58b9a97cb459b85484f compare.csv
100755 blob 30 92c4bc48245c00408cd7e1fd89bc1a03058f4ce4 configure
040000 tree - 8ffdcb3d5e12de55f23f507ed41bfda98d7e9595 debian
040000 tree - a51a4ec84e294b8ba587fda746da7afbf7a60d38 docker
040000 tree - f67d3605efbd6422a8acdd953578991139266391 docs
100755 blob 16711 924262b2f8dbf3bbe02358e7f404175732e970d1 duilder
100644 blob 536 bd0a0092679dbe91dfae0c27f18b536602bd5dd5 duilder.conf
040000 tree - b0cc8cc0386eddf4373339a7860e46e8f74e0202 hooks
040000 tree - 554eb9b51e4a897a00f18b7e20600fd57cfdba5b inc
040000 tree - ab5cc695f620de9abecc84af49866a45612067c6 misc
100644 blob 3890 9bd6ba1151d40eec9064ad66bcc1a9c4cf259e59 rocketgit.spec.in
040000 tree - 07c462d105d023b6a86f8596dd1d8143e9c83128 root
040000 tree - 54c3ef3bee7da1a11d128084874040c0ca6aa709 samples
040000 tree - 7100e75cc1160d3ea84a13b58686d208674ee776 scripts
040000 tree - 00c52dce99b99f5f59800512ffd8e145d5ffe2c9 selinux
100755 blob 256 462ccd108c431f54e380cdac2329129875a318b5 spell_check.sh
040000 tree - cb54e074b3ca35943edfcda9dd9cfcd281bcd9e7 techdocs
040000 tree - 12072c5ed81456a9da14a401a6459e87419434c8 tests
040000 tree - 63f68e921ac8d6a62ea9c3d180e072c7c4725b7d tools
Hints:
Before first commit, do not forget to setup your git environment:
git config --global user.name "your_name_here"
git config --global user.email "your@email_here"

Clone this repository using HTTP(S):
git clone https://rocketgit.com/user/catalinux/rocketgit

Clone this repository using ssh (do not forget to upload a key first):
git clone ssh://rocketgit@ssh.rocketgit.com/user/catalinux/rocketgit

Clone this repository using git:
git clone git://git.rocketgit.com/user/catalinux/rocketgit

You are allowed to anonymously push to this repository.
This means that your pushed commits will automatically be transformed into a merge request:
... clone the repository ...
... make some changes and some commits ...
git push origin main