Marker interface example

src/Adapter/ImmediateProcessingAdapterInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Adapter;
+
+/**
+ * Marker for adapters whose messages must be handled in the same process right after push.
+ */
+interface ImmediateProcessingAdapterInterface {}

src/Adapter/SynchronousAdapter.php

@@ -7,39 +7,15 @@
 use InvalidArgumentException;
 use Yiisoft\Queue\MessageStatus;
 use Yiisoft\Queue\Message\MessageInterface;
-use Yiisoft\Queue\QueueInterface;
-use Yiisoft\Queue\Worker\WorkerInterface;
 use Yiisoft\Queue\Message\IdEnvelope;
 
-use function count;
-
-final class SynchronousAdapter implements AdapterInterface
+final class SynchronousAdapter implements AdapterInterface, ImmediateProcessingAdapterInterface
 {
-    private array $messages = [];
-    private int $current = 0;
-
-    public function __construct(
-        private readonly WorkerInterface $worker,
-        private readonly QueueInterface $queue,
-    ) {}
-
-    public function __destruct()
-    {
-        $this->runExisting(function (MessageInterface $message): bool {
-            $this->worker->process($message, $this->queue);
-
-            return true;
-        });
-    }
+    private int $processed = 0;
 
     public function runExisting(callable $handlerCallback): void
     {
-        $result = true;
-        while (isset($this->messages[$this->current]) && $result === true) {
-            $result = $handlerCallback($this->messages[$this->current]);
-            unset($this->messages[$this->current]);
-            $this->current++;
-        }
+        // Messages are handled immediately in Queue::push().
     }
 
     public function status(string|int $id): MessageStatus
@@ -50,21 +26,17 @@ public function status(string|int $id): MessageStatus
             throw new InvalidArgumentException('This adapter IDs start with 0.');
         }
 
-        if ($id < $this->current) {
+        if ($id < $this->processed) {
             return MessageStatus::DONE;
         }
 
-        if (isset($this->messages[$id])) {
-            return MessageStatus::WAITING;
-        }
-
         throw new InvalidArgumentException('There is no message with the given ID.');
     }
 
     public function push(MessageInterface $message): MessageInterface
     {
-        $key = count($this->messages) + $this->current;
-        $this->messages[] = $message;
+        $key = $this->processed;
+        $this->processed++;
 
         return new IdEnvelope($message, $key);
     }

src/Queue.php

@@ -7,6 +7,7 @@
 use BackedEnum;
 use Psr\Log\LoggerInterface;
 use Yiisoft\Queue\Adapter\AdapterInterface;
+use Yiisoft\Queue\Adapter\ImmediateProcessingAdapterInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Exception\AdapterConfiguration\AdapterNotConfiguredException;
 use Yiisoft\Queue\Message\MessageInterface;
@@ -58,9 +59,9 @@ public function push(
         );
 
         $request = new PushRequest($message, $this->adapter);
-        $message = $this->pushMiddlewareDispatcher
-            ->dispatch($request, $this->createPushHandler(...$middlewareDefinitions))
-            ->getMessage();
+        $request = $this->pushMiddlewareDispatcher
+            ->dispatch($request, $this->createPushHandler(...$middlewareDefinitions));
+        $message = $request->getMessage();
 
         /** @var string $messageId */
         $messageId = $message->getMetadata()[IdEnvelope::MESSAGE_ID_KEY] ?? 'null';
@@ -69,6 +70,10 @@ public function push(
             ['handlerName' => $message->getHandlerName(), 'id' => $messageId],
         );
 
+        if ($request->getAdapter() instanceof ImmediateProcessingAdapterInterface) {
+            $this->handle($message);
+        }
+
         return $message;
     }