xyzshop/rdispatcher/processor.php

<?php

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

class processor extends queue\ILooper
{
    public function __construct()
    {
        parent::__construct(new queue\DispatcherServer());
    }

    protected function handle($msg)
    {
        if(empty($msg)) {
            Log::record('empty body',Log::DEBUG);
        }
        else
        {
            foreach ($msg as $key => $params)
            {
                Log::record("start one", Log::DEBUG);
                $method = strtolower($key);

                $proxy = new proxy();
                if($method == 'add')
                {
                    go(function () use ($proxy,$params) {
                        try {
                            $proxy->add($params);
                        }
                        catch (Exception $x) {
                            Log::record($x->getMessage(),Log::ERR);
                        }
                    });
                }
                elseif($method == 'notify') {

                }
                elseif($method == 'notify_mechant') {

                }
                else {
                    Log::record("Error msg",Log::DEBUG);
                }
            }
        }
    }
}