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))
        {
            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 == 'notify') {
                        $channel = $params['channel'];
                        $input = $params['params'];

                        if(empty($channel) || empty($params))
                            return;
                        $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 == 'query') {
                        $order_id = intval($params['order_id']);
                        $this->mProxy->query($order_id);
                    }
                    else {
                        Log::record("Error Method={$method}", Log::DEBUG);
                    }
                }
                catch (Exception $x) {
                    Log::record($x->getMessage(), Log::ERR);
                }
            }
        }
    }
}