huahuibao/common/components/rabbitmq/notifyConsumer.php

<?php
/**
 * 消息通知消费者
 * 处理各种消息通知：新订单、新客户、充值、销账、配送状态变更等
 */

namespace common\components\rabbitmq;

use biz\wx\classes\WxMessageClass;
use bizGhs\order\classes\OrderClass;
use bizHd\device\classes\HdDeviceClass;
use bizHd\purchase\classes\PurchaseClass;
use bizHd\shop\classes\ShopClass;
use common\components\noticeUtil;
use common\components\push;
use mikemadisonweb\rabbitmq\components\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class notifyConsumer extends baseConsumer
{
    /**
     * 执行消费者逻辑
     *
     * @param AMQPMessage $msg 消息对象
     * @return string 消息处理结果
     *
     * ConsumerInterface::MSG_ACK - 确认消息（标记为已处理）并从队列中删除
     * ConsumerInterface::MSG_REJECT - 拒绝并从队列中删除消息
     * ConsumerInterface::MSG_REQUEUE - 拒绝并重新入队消息
     */
    public function execute(AMQPMessage $msg)
    {
        try {
            $this->ensureDbConnection();
            // 反序列化消息体
            $data = unserialize($msg->body);
            if (!is_array($data)) {
                noticeUtil::push("通知的消费者报错：Invalid notify message format: {$msg->body}", '15280215347');
                return ConsumerInterface::MSG_REJECT;
            }
            print_r($data);
            // 根据通知类型分发处理
            $type = $data['type'] ?? null;
            $result = $this->runWithDbReconnect(function () use ($type, $data) {
                switch ($type) {
                    case 'ghs_new_order_notify':
                        //供货商的新订单通知
                        return $this->ghsNewOrderNotify($data);
                    case 'hd_new_order_notify':
                        //花店的新订单通知
                        return $this->hdNewOrderNotify($data);
                    case 'hd_new_cg_notify':
                        //花店的新采购单通知
                        return $this->hdNewCgNotify($data);
                    case 'ghs_pt_error':
                        //供货商跑腿异常通知
                        return $this->ghsPtErrorAction($data);
                    case 'hd_pt_error':
                        //花店跑腿异常通知
                        return $this->hdPtErrorAction($data);
                    default:
                        noticeUtil::push("通知的消费者提示：Unknown notify type: {$type}", '15280215347');
                        return false;
                }
            });
            if ($result) {
                return ConsumerInterface::MSG_ACK;
            } else {
                noticeUtil::push("通知的消费者提示：Notify message processing failed", '15280215347');
                return ConsumerInterface::MSG_REQUEUE;
            }
        } catch (\Exception $e) {
            noticeUtil::push("Notify consumer exception: " . $e->getMessage(), '15280215347');
            return ConsumerInterface::MSG_REQUEUE;
        }
    }

    private function ghsPtErrorAction($data)
    {
        $orderId = $data['orderId'] ?? 0;
        $event = $data['event'] ?? '';
        $order = OrderClass::getById($orderId, true);
        if (!empty($order)) {
            $sendNum = $order->sendNum ?? '';
            $result = $event . '，取货号 ' . $sendNum;
            $customName = $order->customName ?? '';
            $shopId = $order->shopId ?? '';
            $shop = ShopClass::getById($shopId, true);
            if (!empty($shop)) {
                WxMessageClass::expressErrorInform($shop, $customName, $orderId, $result);
            }
        }
        return true;
    }

    private function hdPtErrorAction($data)
    {
        $orderId = $data['orderId'] ?? 0;
        $event = $data['event'] ?? '';
        $order = \bizHd\order\classes\OrderClass::getById($orderId, true);
        if (!empty($order)) {
            $sendNum = $order->sendNum ?? '';
            $result = $event . '，取货号 ' . $sendNum;
            $customName = $order->customName ?? '';
            $shopId = $order->shopId ?? '';
            $shop = ShopClass::getById($shopId, true);
            if (!empty($shop)) {
                WxMessageClass::expressErrorInform($shop, $customName, $orderId, $result);
            }
        }
        return true;
    }

    private function ghsNewOrderNotify($data)
    {
        $orderId = $data['orderId'] ?? 0;
        if (empty($orderId)) {
            return false;
        }
        $order = OrderClass::getById($orderId, true);
        if (empty($order)) {
            return false;
        }
        $shopId = $order->shopId ?? 0;
        $shop = ShopClass::getById($shopId, true);
        if (empty($shop)) {
            return false;
        }
        if ($shop->actTime != strtotime(date('Y-m-d'))) {
            // actTime 活跃时间，只记录以天为单位的时候戳，用于花店注册时推荐批发店
            $shop->actTime = strtotime(date('Y-m-d'));
            $shop->save();
        }

        WxMessageClass::ghsHasNewOrderInform($shop, $order);

        // 向供货商App发通知
        $cgId = $order->purchaseId ?? 0;
        $cg = PurchaseClass::getById($cgId, true);
        $cgShop = ShopClass::getById($cg->shopId, true, 'shopName, merchantName');
        $push = new push('ghs', push::MSG_TYPE_ORDER);
        $push->ghsOrderMessage($shop, $cgShop, $order);
        return true;
    }

    private function hdNewOrderNotify($data)
    {
        return true;
    }

    private function hdNewCgNotify($data)
    {
        $cgId = $data['cgId'] ?? 0;
        $cg = PurchaseClass::getById($cgId, true);
        $shopId = $cg->shopId ?? 0;
        $shop = ShopClass::getById($shopId, true);
        $allDevices = HdDeviceClass::getAllByCondition(['shopId' => $shopId, 'status' => 1]);
        $cids = array_column($allDevices, 'clientId');
        $title = '买花成功';
        $shopName = $shop->merchantName . ($shop->shopName != '首店' ? '（' . $shop->shopName . '）' : '');
        $content = $shopName . ' ¥' . $cg->actPrice;
        $payload = [
            "page" => "pagesPurchase/purDetails",
            "params" => ["id" => $cgId]
        ];
        $push = new push('hd', push::MSG_TYPE_ORDER);
        $push->pushByCloud($cids, $title, $content, $payload);
        //$str = implode(',', $cids);
        //noticeUtil::push("需要通知：" . $content . ' ' . $str, '15280215347');
        return true;
    }

}