新增消费者弹出、消费前后置事件

Author: Yurunsoft

src/Event/Listener/IConsumerAfterConsumeListener.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Listener;
+
+use Imi\Queue\Event\Param\ConsumerAfterConsumeParam;
+
+/**
+ * 消费者消费消息后置事件
+ */
+interface IConsumerAfterConsumeListener
+{
+    /**
+     * 事件处理方法
+     * @param ConsumerAfterConsumeParam $e
+     * @return void
+     */
+    public function handle(ConsumerAfterConsumeParam $e);
+
+}

src/Event/Listener/IConsumerAfterPopListener.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Listener;
+
+use Imi\Queue\Event\Param\ConsumerAfterPopParam;
+
+/**
+ * 消费者弹出消息后置事件
+ */
+interface IConsumerAfterPopListener
+{
+    /**
+     * 事件处理方法
+     * @param ConsumerAfterPopParam $e
+     * @return void
+     */
+    public function handle(ConsumerAfterPopParam $e);
+
+}

src/Event/Listener/IConsumerBeforeConsumeListener.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Listener;
+
+use Imi\Queue\Event\Param\ConsumerBeforeConsumeParam;
+
+/**
+ * 消费者消费消息前置事件
+ */
+interface IConsumerBeforeConsumeListener
+{
+    /**
+     * 事件处理方法
+     * @param ConsumerBeforeConsumeParam $e
+     * @return void
+     */
+    public function handle(ConsumerBeforeConsumeParam $e);
+
+}

src/Event/Listener/IConsumerBeforePopListener.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Listener;
+
+use Imi\Queue\Event\Param\ConsumerBeforePopParam;
+
+/**
+ * 消费者弹出消息前置事件
+ */
+interface IConsumerBeforePopListener
+{
+    /**
+     * 事件处理方法
+     * @param ConsumerBeforePopParam $e
+     * @return void
+     */
+    public function handle(ConsumerBeforePopParam $e);
+
+}

src/Event/Param/ConsumerAfterConsumeParam.php

@@ -0,0 +1,25 @@
+<?php
+namespace Imi\Queue\Event\Param;
+
+use Imi\Event\EventParam;
+
+/**
+ * 消费者消费消息后置事件参数
+ */
+class ConsumerAfterConsumeParam extends EventParam
+{
+    /**
+     * 队列对象
+     *
+     * @var \Imi\Queue\Driver\IQueueDriver
+     */
+    public $queue;
+
+    /**
+     * 消息
+     *
+     * @var \Imi\Queue\Contract\IMessage
+     */
+    public $message;
+
+}

src/Event/Param/ConsumerAfterPopParam.php

@@ -0,0 +1,25 @@
+<?php
+namespace Imi\Queue\Event\Param;
+
+use Imi\Event\EventParam;
+
+/**
+ * 消费者弹出消息后置事件参数
+ */
+class ConsumerAfterPopParam extends EventParam
+{
+    /**
+     * 队列对象
+     *
+     * @var \Imi\Queue\Driver\IQueueDriver
+     */
+    public $queue;
+
+    /**
+     * 消息
+     *
+     * @var \Imi\Queue\Contract\IMessage
+     */
+    public $message;
+
+}

src/Event/Param/ConsumerBeforeConsumeParam.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Param;
+
+use Imi\Event\EventParam;
+
+/**
+ * 消费者消费消息前置事件参数
+ */
+class ConsumerBeforeConsumeParam extends EventParam
+{
+    /**
+     * 队列对象
+     *
+     * @var \Imi\Queue\Driver\IQueueDriver
+     */
+    public $queue;
+
+}

src/Event/Param/ConsumerBeforePopParam.php

@@ -0,0 +1,18 @@
+<?php
+namespace Imi\Queue\Event\Param;
+
+use Imi\Event\EventParam;
+
+/**
+ * 消费者弹出消息前置事件参数
+ */
+class ConsumerBeforePopParam extends EventParam
+{
+    /**
+     * 队列对象
+     *
+     * @var \Imi\Queue\Driver\IQueueDriver
+     */
+    public $queue;
+
+}

src/Service/BaseQueueConsumer.php

@@ -1,13 +1,18 @@
 <?php
 namespace Imi\Queue\Service;
 
+use Imi\Event\Event;
 use Swoole\Coroutine;
 use Imi\Aop\Annotation\Inject;
 use Yurun\Swoole\CoPool\CoPool;
 use Imi\Queue\Contract\IMessage;
 use Imi\Queue\Driver\IQueueDriver;
 use Yurun\Swoole\CoPool\Interfaces\ICoTask;
 use Yurun\Swoole\CoPool\Interfaces\ITaskParam;
+use Imi\Queue\Event\Param\ConsumerAfterPopParam;
+use Imi\Queue\Event\Param\ConsumerBeforePopParam;
+use Imi\Queue\Event\Param\ConsumerAfterConsumeParam;
+use Imi\Queue\Event\Param\ConsumerBeforeConsumeParam;
 
 /**
  * 队列消费基类
@@ -64,14 +69,28 @@ public function start(?int $co = null)
         $task = function() use($config){
             $queue = $this->imiQueue->getQueue($this->name);
             do {
+                Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_POP', [
+                    'queue' =>  $queue,
+                ], $this, ConsumerBeforePopParam::class);
                 $message = $queue->pop();
+                Event::trigger('IMI.QUEUE.CONSUMER.AFTER_POP', [
+                    'queue'     =>  $queue,
+                    'message'   =>  $message,
+                ], $this, ConsumerAfterPopParam::class);
                 if(null === $message)
                 {
                     Coroutine::sleep($config->getTimespan());
                 }
                 else
                 {
+                    Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_CONSUME', [
+                        'queue' =>  $queue,
+                    ], $this, ConsumerBeforeConsumeParam::class);
                     $this->consume($message, $queue);
+                    Event::trigger('IMI.QUEUE.CONSUMER.AFTER_CONSUME', [
+                        'queue'     =>  $queue,
+                        'message'   =>  $message,
+                    ], $this, ConsumerAfterConsumeParam::class);
                 }
             } while($this->working);  
         };