xyzshop/rdispatcher/processor.php

<?php

require_once(BASE_CORE_PATH . '/framework/function/http.php');

class processor extends queue\ILooper
{
    private $mProxy;

    public function __construct($comode = false)
    {
        if ($comode) {
            parent::__construct(new queue\CoDispatcherServer());
        } else {
            parent::__construct(new queue\DispatcherServer());
        }
        $this->mProxy = new proxy();
    }

    protected function handle($msg)
    {
        if (empty($msg)) return '';

        $method = '';
        foreach ($msg as $key => $params)
        {
            if (empty($params)) continue;
            $method = strtolower($key);
            Log::record("processor hanlde method=$method", Log::DEBUG);
            try
            {
                if ($method == 'add') {
                    $this->mProxy->add($params);
                } elseif ($method == 'add_zero') {
                    $this->mProxy->add_zero($params);
                } elseif ($method == 'notify') {
                    $channel = $params['channel'];
                    $input = $params['params'];
                    if (empty($channel) || empty($params))
                        continue;
                    $this->mProxy->notify($channel, $input);
                } elseif ($method == 'notify_mechant') {
                    $order_id = intval($params['order_id']);
                    $manual = $params['manual'] ?? false;
                    $this->mProxy->notify_merchant($order_id, $manual);
                } elseif ($method == 'notify_merchant_success') { //发预回调通知.
                    $mchid = intval($params['mchid']);
                    $mch_order = trim($params['mch_order']);
                    $this->mProxy->notify_merchant_success($mchid, $mch_order);
                } elseif ($method == 'query') {
                    $order_id = intval($params['order_id']);
                    $this->mProxy->query($order_id);
                } elseif ($method == 'query_auto') {
                    $order_id = intval($params['order_id']);
                    $query_times = intval($params['query_times']);
                    $this->mProxy->query_auto($order_id,$query_times);
                } elseif ($method == 'query_net') {
                    $order_id = intval($params['order_id']);
                    $this->mProxy->query_net($order_id);
                } elseif ($method == 'query_timeout') {
                    $mchid = intval($params['mchid']);
                    $mch_order = trim($params['mch_order']);
                    $this->mProxy->query_timeout($mchid,$mch_order);
                } elseif ($method == 'manual_success') {
                    $order_id = intval($params['order_id']);
                    $this->mProxy->manual_success($order_id);
                } elseif ($method == 'manual_cancel') {
                    $order_id = intval($params['order_id']);
                    $this->mProxy->manual_cancel($order_id);
                } elseif ($method == 'addthird') {
                    $this->mProxy->addthird($params);
                } else {
                    Log::record("Error Method={$method}", Log::DEBUG);
                }
            }
            catch (Exception $x) {
                Log::record("processor::handle ". $x->getMessage(), Log::ERR);
            }
        }

        return $method;
    }
}