1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

<event>
    <handle>
        <order_invoice>
            <event_data_initialization>
                <handle>OrderInvoiceInitHandle</handle>
                <sort_order>0</sort_order>
            </event_data_initialization>
            <reward_points_balance_update>
                <handle>IncreaseRewardPointsBalanceHandle</handle>
                <sort_order>10</sort_order>
            </reward_points_balance_update>
            <customer_upgrade_group>
                <handle>CustomerGroupUpdateHandle</handle>
                <sort_order>20</sort_order>
            </customer_upgrade_group>
            <reward_points_update_notification>
                <handle>SendRewardPointsNotificationHandle</handle>
                <sort_order>100</sort_order>
            </reward_points_update_notification>
        </order_invoice>
    </handle>
</event>

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

class EventDispatcher
{
    const XML_PATH_EVENT_HANDLES_PREFIX = 'event/handle/';

    protected $logger;
    protected $objectManager;
    protected $scopeConfig;

    public function __construct(
        LoggerInterface $logger,
        ScopeConfigInterface $scopeConfig,
        ObjectManagerInterface $objectManager
    ) {
        $this->logger = $logger;
        $this->objectManager = $objectManager;
        $this->scopeConfig = $scopeConfig;
    }


    public function execute()
    {
        try {
            /** @var EventQueue $eventModel */
            $eventModel = $this->objectManager->create(EventQueue::class);
            $conn = $eventModel->getResource()->getConnection();
            $select = $conn->select();
            $select->reset()
                ->from($conn->getTableName('oneday_event_queue'), ['id'])
                ->where('status = ?', EventQueue::STATUS_PENDING)
                ->where('scheduled_at < ?', date('Y-m-d H:i:s', time() + 60));

            $stmt = $select->query();

            /** @var EventQueue $event */
            while ($row = $stmt->fetch()) {
                try {
                    $event = clone $eventModel;
                    $event->load($row['id']);
                    if ($event->getStatus() !== EventQueue::STATUS_PENDING) {
                        continue;
                    }
                    // 设置 Event 为 running 状态
                    $event->setStatus(EventQueue::STATUS_RUNNING)
                        ->setStartedAt(date('Y-m-d H:i:s'));
                    $event->save();

                    $handles = $this->getHandlesByCode($event->getCode());
                    $messages = [];
                    /** @var DataObject $eventData */
                    $eventData = $this->objectManager->create(DataObject::class);
                    $eventData->addData(json_decode($event->getEventData(), true));
                    foreach ($handles as $handle) {
                        // 开始处理 Event
                        try {
                            /** @var EventExecutorInterface $handleInstance */
                            $handleInstance = $this->objectManager->create($handle);
                            if ($handleInstance instanceof EventExecutorInterface) {
                                $handleInstance->execute($eventData);
                            } else {
                                $messages[] = 'Event Handle('. $handle. ') have not implements '. EventExecutorInterface::class. 'Skipped this event handle';
                            }
                        } catch (\Exception $e) {
                            $messages[] = 'Event execute exception: '. $e->getMessage();
                        }
                    }
                    // Event 处理完成, 设置为 finished 状态
                    $event->setFinishedAt(time())
                        ->setStatus(EventQueue::STATUS_FINISHED)
                        ->setMessage(implode("\n", $messages));
                    $event->save();
                } catch (\Exception $e) {
                    // Event 处理失败, 设置为 error 状态
                    // todo 改进处理失败方式
                    $message = '';
                    if (isset($messages) && count($messages)) {
                        $message = implode("\n", $messages);
                    }
                    $event->setMessage($message. "\n". $e->getMessage())
                        ->setStatus(EventQueue::STATUS_ERROR);
                    $event->save();
                    $this->logger->critical(sprintf('processing event#%s failed, %s', $event->getId(), $e->getMessage()));
                }
            }
        } catch (\Exception $e) {
            $this->logger->critical('Event process error: '. $e->getMessage());
        }
    }

    /**
     * 根据 code 获取配置的 event handle
     * @param $code
     * @return array
     */
    protected function getHandlesByCode($code)
    {
        $configData = $this->scopeConfig->getValue(self::XML_PATH_EVENT_HANDLES_PREFIX. $code);
        $handles = [];
        if (is_array($configData)) {
            $_handles = [];
            foreach ($configData as $item) {
                if (isset($item['handle']) && $item['handle']) {
                    $sortOrder = isset($item['sort_order']) ? (int) $item['sort_order'] : 1;
                    $handle = $item['handle'];
                    if (isset($_handles[$sortOrder])) {
                        $_handles[$sortOrder][] = $handle;
                    } else {
                        $_handles[$sortOrder] = [$handle];
                    }
                }
            }
            ksort($_handles);
            foreach ($_handles as $items) {
                $handles = array_merge($handles, $items);
            }
        }
        return $handles;
    }
}