Processed timed-out Rules before handled Rules, ensuring that we don't call fire() twice for a timed-out Rule. Was noticeable with Cron shutdown onSchedule firing twice.

Author: lucasnetau

src/CorrelationEngine.php

@@ -173,7 +173,7 @@ public function handle(Event $event) : void
             $expecting = $matcher->nextAcceptedEvents();
             $result = $matcher->handle($event);
             if ($this->isFlagSet($result, $matcher::EVENT_HANDLED)) {
-                $handledMatchers[] = $matcher;
+                $handledMatchers[spl_object_id($matcher)] = $matcher;
                 $skipMatchers[] = get_class($matcher);
                 /** Update which events we are expecting next **/
                 $this->removeWatchForEvents($matcher, $expecting);
@@ -186,7 +186,7 @@ public function handle(Event $event) : void
 
             // If we are timed out then flag the matcher for timeout processing.
             if ($this->isFlagSet($result, $matcher::EVENT_TIMEOUT)) {
-                $timedOutMatchers[] = $matcher;
+                $timedOutMatchers[spl_object_id($matcher)] = $matcher;
             }
 
             //Matcher has told us to suppress further processing of this event.
@@ -214,7 +214,7 @@ public function handle(Event $event) : void
                 $matcher = $this->constructMatcher($class);
                 $result = $matcher->handle($event);
                 if ($this->isFlagSet($result, $matcher::EVENT_HANDLED)) {
-                    $handledMatchers[] = $matcher;
+                    $handledMatchers[spl_object_id($matcher)] = $matcher;
                     $this->eventProcessors[spl_object_id($matcher)] = $matcher;
                     $this->incrStat('init_matcher', $class);
                     $this->incrStat('handled', (string)$event->event . "|" . get_class($matcher));
@@ -241,8 +241,21 @@ public function handle(Event $event) : void
             $this->incrStat('unhandled', (string)$event->event);
         }
 
-        /** For any matchers that processed this event fire any actions, then update timeout or destroy if complete **/
         $stateChanged = !(empty($handledMatchers) && empty($timedOutMatchers));
+
+        /**  Fire any action and destroy any timed out matchers **/
+        foreach ($timedOutMatchers as $objectId => $matcher) {
+            $this->removeTimeout($matcher);
+            $matcher->fire();
+            /** Record stat of matcher timeout */
+            $this->incrStat('completed_matcher_timeout', get_class($matcher));
+            $this->removeMatcher($matcher);
+            unset($handledMatchers[$objectId]);
+            unset($matcher);
+        }
+        unset($timedOutMatchers);
+
+        /** For any matchers that processed this event fire any actions, then update timeout or destroy if complete **/
         foreach ($handledMatchers as $matcher) {
             $matcher->fire();
             $this->addTimeout($matcher);
@@ -254,16 +267,7 @@ public function handle(Event $event) : void
                 unset($matcher);
             }
         }
-
-        /**  Fire any action and destroy any timed out matchers **/
-        foreach ($timedOutMatchers as $matcher) {
-            $this->removeTimeout($matcher);
-            $matcher->fire();
-            /** Record stat of matcher timeout */
-            $this->incrStat('completed_matcher_timeout', get_class($matcher));
-            $this->removeMatcher($matcher);
-            unset($matcher);
-        }
+        unset($handledMatchers);
 
         /** Flag as dirty except when we have a control message and state doesn't change **/
         if ($stateChanged || !in_array($event->event, Scheduler::CONTROL_MESSAGES, true)) {

tests/cron_timer.php

@@ -0,0 +1,101 @@
+<?php declare(strict_types=1);
+
+use EdgeTelemetrics\EventCorrelation\Event;
+use EdgeTelemetrics\EventCorrelation\Rule;
+use Bref\Logger\StderrLogger;
+use EdgeTelemetrics\EventCorrelation\Scheduler;
+
+use Psr\Log\LogLevel;
+use function EdgeTelemetrics\EventCorrelation\php_cmd;
+use function EdgeTelemetrics\EventCorrelation\wrap_source_php_cmd;
+
+error_reporting(E_ALL);
+ini_set('display_errors', "on");
+
+include __DIR__ . "/../vendor/autoload.php";
+
+class EveryMinuteCron extends Rule\Cron {
+
+    const TIMEZONE = 'Australia/Sydney';
+    const CRON_SCHEDULE = '* * * * *';
+
+    public function onSchedule(): void
+    {
+        error_log('MINUTE CRON: Minute Cron schedule reached ' . (new DateTimeImmutable())->format('Y-m-d H:i:s'));
+    }
+}
+
+class InitCron extends Rule\Cron {
+    const CRON_SCHEDULE = Rule\Cron::ON_INITIALISATION;
+
+    public function onSchedule(): void
+    {
+        error_log('INIT CRON: System has initialised');
+    }
+}
+
+class ShutdownCron extends Rule\Cron {
+    const CRON_SCHEDULE = Rule\Cron::ON_SHUTDOWN;
+
+    public function onSchedule(): void
+    {
+        error_log('SHUTDOWN CRON: System shutting down');
+    }
+}
+
+$numberGenClass = new class() extends Scheduler\SourceFunction {
+    protected \React\EventLoop\TimerInterface $timer;
+
+    function functionStart(): void
+    {
+        $this->timer = $this->loop->addPeriodicTimer(5.0, function () {
+            static $count = 1;
+            try {
+                $event = new Event(['event' => 'SampleValueEvent', 'value' => $count++]);
+                $this->emit('data', [$event]);
+            } catch (Throwable $exception) {
+                $this->emit('error', [$exception]);
+                $this->exit(255);
+                return;
+            }
+        });
+    }
+
+    function exit(int $code = 0): void
+    {
+        $this->running = false;
+        $this->loop->cancelTimer($this->timer);
+        unset($this->timer);
+        $this->emit('exit', [$code]);
+    }
+
+    function functionStop(): void
+    {
+        $this->exit();
+    }
+};
+
+$scheduler = new class([EveryMinuteCron::class, InitCron::class, ShutdownCron::class]) extends Scheduler {
+    public function __construct(array $rules)
+    {
+        parent::__construct($rules);
+        set_exception_handler([$this, "handle_exception"]);
+        $this->setLogger(new StderrLogger(LogLevel::DEBUG));
+
+        //$this->engine->setEventStreamLive();
+
+        $this->setSavefileName("/tmp/php_ec-scheduler_test.state");
+        $this->setSaveStateInterval(1);
+        $this->enableManagementServer(true);
+        $this->setHeartbeatInterval(10);
+    }
+
+    function handle_exception($exception): void
+    {
+        $this->logger->emergency("Fatal", ['exception' => $exception,]);
+    }
+};
+
+$scheduler->register_input_process('generator', $numberGenClass, null, [], false);
+
+$scheduler->run();