Stop ack-ing unprocessed pubsub messages and log warnings

composer.json

@@ -6,26 +6,30 @@
         {
             "name": "Leo Sjöberg",
             "email": "[email protected]"
+        },
+        {
+            "name": "Hakan Aktas",
+            "email": "[email protected]"
         }
     ],
     "require": {
-        "php": ">= 7.3",
+        "php": ">=8.2",
         "google/cloud": "0.*",
-        "illuminate/config": "^6.0|^7.0|^8.0|^9.0|^10.0",
-        "illuminate/console": "^6.0|^7.0|^8.0|^9.0|^10.0",
+        "illuminate/config": "^9.0|^10.0",
+        "illuminate/console": "^9.0|^10.0",
         "kainxspirits/laravel-pubsub-queue": "~0.4.0|~0.5.0|~0.6.0|~0.7.0|~0.8.0",
         "google/protobuf": "^3.6.0",
-        "illuminate/bus": "^6.0|^7.0|^8.0|^9.0|^10.0",
+        "illuminate/bus": "^9.0|^10.0",
         "ext-json": "*"
     },
     "autoload": {
         "psr-4": {
-            "Decahedron\\AppEvents\\": "src/"
+            "Jobilla\\AppEvents\\": "src/"
         }
     },
     "autoload-dev": {
         "psr-4": {
-            "Decahedron\\AppEvents\\Tests\\Proto\\": "tests/fixtures/Decahedron/AppEvents/Tests/Proto"
+            "Jobilla\\AppEvents\\Tests\\Proto\\": "tests/fixtures/Jobilla/AppEvents/Tests/Proto"
         }
     },
     "extra": {
@@ -34,7 +38,7 @@
                 "Kainxspirits\\PubSubQueue\\PubSubQueueServiceProvider"
             ],
             "providers": [
-                "Decahedron\\AppEvents\\AppEventsProvider"
+                "Jobilla\\AppEvents\\AppEventsProvider"
             ]
         }
     },

src/AppEvent.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Illuminate\Bus\Queueable;
 use Illuminate\Container\Container;
@@ -34,15 +34,4 @@ public function __construct(string $event, Message $payload)
         $this->payload = $payload;
         $this->event = $event;
     }
-
-    public function handle()
-    {
-        foreach (Config::get('app-events.handlers') as $event => $handler) {
-            if ($this->event !== $event) {
-                continue;
-            }
-
-            Container::getInstance()->make($handler)->handle($this->payload, $event);
-        }
-    }
 }

src/AppEventFactory.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Google\Cloud\PubSub\Message;
 use Google\Protobuf\Internal\Message as ProtobufMessage;
@@ -12,9 +12,25 @@ class AppEventFactory
      * @throws UnserializableProtoException
      * @throws UnsupportedEventException
      */
-    public static function fromMessage(Message $message): AppEvent
+    public static function fromMessage(Message $message): mixed
     {
-        return new AppEvent($message->attribute('event_type'), static::resolveProtobufInstance($message));
+        $handler = static::resolveHandler($message->attribute('event_type'));
+
+        return new $handler(static::resolveProtobufInstance($message));
+    }
+
+    /**
+     * @throws UnsupportedEventException
+     */
+    protected static function resolveHandler(string $eventName): string
+    {
+        $configKey = 'app-events.handlers.' . $eventName;
+
+        if (! Config::has($configKey)) {
+            throw new UnsupportedEventException($eventName);
+        }
+
+        return Config::get($configKey);
     }
 
     /**

src/AppEventsProvider.php

@@ -1,8 +1,8 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
-use Decahedron\AppEvents\Commands\AppEventsListener;
+use Jobilla\AppEvents\Commands\AppEventsListener;
 use Google\Cloud\PubSub\PubSubClient;
 use Illuminate\Support\ServiceProvider;
 

src/Commands/AppEventsListener.php

@@ -1,18 +1,18 @@
 <?php
 
-namespace Decahedron\AppEvents\Commands;
+namespace Jobilla\AppEvents\Commands;
 
-use Decahedron\AppEvents\UnsupportedEventException;
+use Jobilla\AppEvents\UnsupportedEventException;
 use Exception;
 use Google\Cloud\Core\Exception\BadRequestException;
 use Google\Cloud\PubSub\Subscription;
 use Illuminate\Console\Command;
 use Illuminate\Support\Facades\Log;
 use Google\Cloud\PubSub\PubSubClient;
-use Decahedron\AppEvents\AppEventFactory;
+use Jobilla\AppEvents\AppEventFactory;
 use Illuminate\Contracts\Config\Repository;
-use Decahedron\AppEvents\UnserializableProtoException;
-use Decahedron\AppEvents\SubscriptionTopicMismatchException;
+use Jobilla\AppEvents\UnserializableProtoException;
+use Jobilla\AppEvents\SubscriptionTopicMismatchException;
 
 class AppEventsListener extends Command
 {
@@ -101,9 +101,6 @@ private function runIteration(Subscription $subscription)
             try {
                 $job = AppEventFactory::fromMessage($message);
             } catch (UnserializableProtoException $e) {
-                if (! $this->option('silent')) {
-                    $this->info('No implementation registered for message type: ' . $e->protoMessageType);
-                }
                 $handledMessages[] = $message;
                 continue;
             } catch (UnsupportedEventException $e) {
@@ -112,9 +109,9 @@ private function runIteration(Subscription $subscription)
                     'message' => $message->info(),
                     'published_at' => $message->publishTime(),
                 ]);
-                $handledMessages[] = $message;
                 continue;
             }
+
             if (! $this->option('silent')) {
                 $this->info('Handling message: '.$job->event);
             }

src/PubSubConnector.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Google\Cloud\PubSub\PubSubClient;
 use Kainxspirits\PubSubQueue\Connectors\PubSubConnector as BaseConnector;

src/PubSubQueue.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Google\Cloud\PubSub\Message;
 use Kainxspirits\PubSubQueue\Jobs\PubSubJob;

src/SubscriptionTopicMismatchException.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Exception;
 use Google\Cloud\PubSub\Subscription;

src/UnserializableProtoException.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Exception;
 

src/UnsupportedEventException.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Decahedron\AppEvents;
+namespace Jobilla\AppEvents;
 
 use Exception;
 

tests/AppEventsListenerTest.php

@@ -1,9 +1,9 @@
 <?php
 
-namespace Decahedron\AppEvents\Tests;
+namespace Jobilla\AppEvents\Tests;
 
-use Decahedron\AppEvents\Commands\AppEventsListener;
-use Decahedron\AppEvents\Tests\Proto\Test;
+use Jobilla\AppEvents\Commands\AppEventsListener;
+use Jobilla\AppEvents\Tests\Proto\Test;
 use Google\Cloud\PubSub\Message;
 use Google\Cloud\PubSub\PubSubClient;
 use Google\Cloud\PubSub\Subscription;
@@ -108,4 +108,4 @@ public function handle()
     {
         throw new \Exception('failure');
     }
-}
+}

tests/PubSubQueueTest.php

@@ -1,10 +1,10 @@
 <?php
 
-namespace Decahedron\AppEvents\Tests;
+namespace Jobilla\AppEvents\Tests;
 
-use Decahedron\AppEvents\AppEvent;
-use Decahedron\AppEvents\PubSubQueue;
-use Decahedron\AppEvents\Tests\Proto\Test;
+use Jobilla\AppEvents\AppEvent;
+use Jobilla\AppEvents\PubSubQueue;
+use Jobilla\AppEvents\Tests\Proto\Test;
 use Google\Cloud\PubSub\Message;
 use Google\Cloud\PubSub\PubSubClient;
 use Google\Cloud\PubSub\Subscription;