Responsible for loading workers

@param        $worker
@param array  $params
@param string $method_name
@param array  $data
@return object|string
509593
public function worker($worker, $params = array(), $method_name = "", $data = array())
    {
        //Check if this worker has been included
        if (!is_array($worker)) {
            $worker = array($worker);
        }
        $helper = "";
        foreach ($worker as $helper) {

            $helper = ucfirst($helper);
            $file = configItem('site')['workerdir'] . $helper .'/'. $helper . '.php';

            $class = ucfirst(basename($file, '.php'));

            if (!file_exists($file)) {

                $this->logger->error("The worker $file was not found");

            }
            $this->loading[] = 'worker';
            $helper = $this->$class = $this->_createInstance('Eaglehorn\worker\\'.$helper.'\\', $class, $params, $method_name, $data);
        }

        return $helper;
    }
273735
protected function upload_worker($worker_name, $worker_file_name)
    {
        $this->info("<info>Starting to upload <comment>$worker_name</comment> worker</info>");
        @$this->worker->upload(getcwd(), $this->workers_dir . '/' . $worker_file_name, $worker_name, $this->worker_params);
        $this->info("<info>Worker <comment>$worker_name</comment> uploaded</info>" . PHP_EOL);
    }
1187014
public function setTask(AbstractWorker $worker, $data): void
    {
        $key = sprintf('worker:%s', $worker->getId());

        $this->client->set($key, json_encode($data));
    }
843881
function WorkerPouch(opts, callback) {
  var api = this;

  if (typeof opts === 'string') {
    var slashIdx = utils.lastIndexOf(opts, '/');
    opts = {
      url: opts.substring(0, slashIdx),
      name: opts.substring(slashIdx + 1)
    };
  } else {
    opts = utils.clone(opts);
  }

  log('constructor called', opts);

  // Aspirational. once https://github.com/pouchdb/pouchdb/issues/5200
  // is resolved, you'll be able to directly pass in a worker here instead of
  // a function that returns a worker.
  var worker = (opts.worker && typeof opts.worker === 'function') ?
    opts.worker() : opts.worker;
  if (!worker || (!worker.postMessage && (!worker.controller || !worker.controller.postMessage))) {
    var workerOptsErrMessage =
      'Error: you must provide a valid `worker` in `new PouchDB()`';
    console.error(workerOptsErrMessage);
    return callback(new Error(workerOptsErrMessage));
  }

  if (!opts.name) {
    var optsErrMessage = 'Error: you must provide a database name.';
    console.error(optsErrMessage);
    return callback(new Error(optsErrMessage));
  }

  function handleUncaughtError(content) {
    try {
      api.emit('error', content);
    } catch (err) {
      // TODO: it's weird that adapters should have to handle this themselves
      console.error(
        'The user\'s map/reduce function threw an uncaught error.\n' +
        'You can debug this error by doing:\n' +
        'myDatabase.on(\'error\', function (err) { debugger; });\n' +
        'Please double-check your map/reduce function.');
      console.error(content);
    }
  }

  function onReceiveMessage(message) {
    var messageId = message.messageId;
    var messageType = message.type;
    var content = message.content;

    if (messageType === 'uncaughtError') {
      handleUncaughtError(content);
      return;
    }

    var cb = api._callbacks[messageId];

    if (!cb) {
      log('duplicate message (ignoring)', messageId, messageType, content);
      return;
    }

    log('receive message', api._instanceId, messageId, messageType, content);

    if (messageType === 'error') {
      delete api._callbacks[messageId];
      cb(content);
    } else if (messageType === 'success') {
      delete api._callbacks[messageId];
      cb(null, content);
    } else { // 'update'
      api._changesListeners[messageId](content);
    }
  }

  function workerListener(e) {
    if (e.data.id === api._instanceId) {
      onReceiveMessage(e.data);
    }
  }

  function postMessage(message) {
    /* istanbul ignore if */
    if (typeof worker.controller !== 'undefined') {
      // service worker, use MessageChannels because e.source is broken in Chrome < 51:
      // https://bugs.chromium.org/p/chromium/issues/detail?id=543198
      var channel = new MessageChannel();
      channel.port1.onmessage = workerListener;
      worker.controller.postMessage(message, [channel.port2]);
    } else {
      // web worker
      worker.postMessage(message);
    }
  }

  function sendMessage(type, args, callback) {
    if (api._destroyed) {
      return callback(new Error('this db was destroyed'));
    } else if (api._closed) {
      return callback(new Error('this db was closed'));
    }
    var messageId = uuid();
    log('send message', api._instanceId, messageId, type, args);
    api._callbacks[messageId] = callback;
    var encodedArgs = encodeArgs(args);
    postMessage({
      id: api._instanceId,
      type: type,
      messageId: messageId,
      args: encodedArgs
    });
    log('message sent', api._instanceId, messageId);
  }

  function sendRawMessage(messageId, type, args) {
    log('send message', api._instanceId, messageId, type, args);
    var encodedArgs = encodeArgs(args);
    postMessage({
      id: api._instanceId,
      type: type,
      messageId: messageId,
      args: encodedArgs
    });
    log('message sent', api._instanceId, messageId);
  }

  api.type = function () {
    return 'worker';
  };

  api._remote = false;

  api._id = adapterFun('id', function (callback) {
    sendMessage('id', [], callback);
  });

  api.compact = adapterFun('compact', function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('compact', [opts], callback);
  });

  api._info = function (callback) {
    sendMessage('info', [], callback);
  };

  api.get = adapterFun('get', function (id, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('get', [id, opts], callback);
  });

  // hacky code necessary due to implicit breaking change in
  // https://github.com/pouchdb/pouchdb/commits/0ddeae6b
  api._get = function (id, opts, callback) {
    api.get(id, opts, function (err, doc) {
      if (err) {
        return callback(err);
      }
      callback(null, {doc: doc});
    });
  };

  api.remove =
    adapterFun('remove', function (docOrId, optsOrRev, opts, callback) {
      var doc;
      if (typeof optsOrRev === 'string') {
        // id, rev, opts, callback style
        doc = {
          _id: docOrId,
          _rev: optsOrRev
        };
        if (typeof opts === 'function') {
          callback = opts;
          opts = {};
        }
      } else {
        // doc, opts, callback style
        doc = docOrId;
        if (typeof optsOrRev === 'function') {
          callback = optsOrRev;
          opts = {};
        } else {
          callback = opts;
          opts = optsOrRev;
        }
      }
      var rev = (doc._rev || opts.rev);

      sendMessage('remove', [doc._id, rev], callback);
  });

  api.getAttachment =
    adapterFun('getAttachment', function (docId, attachmentId, opts,
                                                callback) {
      if (typeof opts === 'function') {
        callback = opts;
        opts = {};
      }
      sendMessage('getAttachment', [docId, attachmentId, opts], callback);
  });

  api.removeAttachment =
    adapterFun('removeAttachment', function (docId, attachmentId, rev,
                                                   callback) {

      sendMessage('removeAttachment', [docId, attachmentId, rev], callback);
    });

  // Add the attachment given by blob and its contentType property
  // to the document with the given id, the revision given by rev, and
  // add it to the database given by host.
  api.putAttachment =
    adapterFun('putAttachment', function (docId, attachmentId, rev, blob,
                                                type, callback) {
      if (typeof type === 'function') {
        callback = type;
        type = blob;
        blob = rev;
        rev = null;
      }
      if (typeof type === 'undefined') {
        type = blob;
        blob = rev;
        rev = null;
      }

      if (typeof blob === 'string') {
        var binary;
        try {
          binary = atob(blob);
        } catch (err) {
          // it's not base64-encoded, so throw error
          return callback(errors.error(errors.BAD_ARG,
            'Attachments need to be base64 encoded'));
        }
        blob = utils.createBlob([utils.binaryStringToArrayBuffer(binary)], {type: type});
      }

      var args = [docId, attachmentId, rev, blob, type];
      sendMessage('putAttachment', args, callback);
    });

  api.put = adapterFun('put', utils.getArguments(function (args) {
    var temp, temptype, opts;
    var doc = args.shift();
    var id = '_id' in doc;
    var callback = args.pop();
    if (typeof doc !== 'object' || Array.isArray(doc)) {
      return callback(errors.error(errors.NOT_AN_OBJECT));
    }

    doc = utils.clone(doc);

    preprocessAttachments(doc).then(function () {
      while (true) {
        temp = args.shift();
        temptype = typeof temp;
        if (temptype === "string" && !id) {
          doc._id = temp;
          id = true;
        } else if (temptype === "string" && id && !('_rev' in doc)) {
          doc._rev = temp;
        } else if (temptype === "object") {
          opts = utils.clone(temp);
        }
        if (!args.length) {
          break;
        }
      }
      opts = opts || {};

      sendMessage('put', [doc, opts], callback);
    }).catch(callback);

  }));

  api.post = adapterFun('post', function (doc, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    opts = utils.clone(opts);

    sendMessage('post', [doc, opts], callback);
  });

  api._bulkDocs = function (req, opts, callback) {
    sendMessage('bulkDocs', [req, opts], callback);
  };

  api._allDocs = function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('allDocs', [opts], callback);
  };

  api._changes = function (opts) {
    opts = utils.clone(opts);

    if (opts.continuous) {
      var messageId = uuid();
      api._changesListeners[messageId] = opts.onChange;
      api._callbacks[messageId] = opts.complete;
      sendRawMessage(messageId, 'liveChanges', [opts]);
      return {
        cancel: function () {
          sendRawMessage(messageId, 'cancelChanges', []);
        }
      };
    }

    sendMessage('changes', [opts], function (err, res) {
      if (err) {
        opts.complete(err);
        return callback(err);
      }
      res.results.forEach(function (change) {
        opts.onChange(change);
      });
      if (opts.returnDocs === false || opts.return_docs === false) {
        res.results = [];
      }
      opts.complete(null, res);
    });
  };

  // Given a set of document/revision IDs (given by req), tets the subset of
  // those that do NOT correspond to revisions stored in the database.
  // See http://wiki.apache.org/couchdb/HttpPostRevsDiff
  api.revsDiff = adapterFun('revsDiff', function (req, opts, callback) {
    // If no options were given, set the callback to be the second parameter
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }

    sendMessage('revsDiff', [req, opts], callback);
  });

  api._query = adapterFun('query', function (fun, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    var funEncoded = fun;
    if (typeof fun === 'function') {
      funEncoded = {map: fun};
    }
    sendMessage('query', [funEncoded, opts], callback);
  });

  api._viewCleanup = adapterFun('viewCleanup', function (callback) {
    sendMessage('viewCleanup', [], callback);
  });

  api._close = function (callback) {
    api._closed = true;
    callback();
  };

  api.destroy = adapterFun('destroy', function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('destroy', [], function (err, res) {
      if (err) {
        api.emit('error', err);
        return callback(err);
      }
      api._destroyed = true;
      worker.removeEventListener('message', workerListener);
      api.emit('destroyed');
      callback(null, res);
    });
  });

  // api.name was added in pouchdb 6.0.0
  api._instanceId = api.name || opts.originalName;
  api._callbacks = {};
  api._changesListeners = {};

  worker.addEventListener('message', workerListener);

  var workerOpts = {
    name: api._instanceId,
    auto_compaction: !!opts.auto_compaction,
    storage: opts.storage
  };
  if (opts.revs_limit) {
    workerOpts.revs_limit = opts.revs_limit;
  }

  sendMessage('createDatabase', [workerOpts], function (err) {
    if (err) {
      return callback(err);
    }
    callback(null, api);
  });
}
1154635
function _messageWorker(message, callback) {
    const worker = _getWorker();

    function _listen(e) {
        if (e.data.id === message.id) {
            callback(e.data);
            worker.removeEventListener('message', _listen);
        }
    }

    worker.addEventListener('message', _listen);
    worker.postMessage(message);
}
1162740
public function dispathToWorkerId($workerId, $type, $data = null)
    {
        get_instance()->sendToOneWorker($workerId, SwooleMarco::DISPATCHER_NAME, [$type, $data], EventDispatcher::class . "::workerDispatchEventWith");
    }
1018531
public function onWorkerStart(\Swoole\Server $server, int $worker_id)
    {
        // STDOUT_LOG模式，不打印 worker stop 输出
        if ( getenv('STDOUT_LOG') ) {
            echo date('Y-m-d H:i:s') . " |\033[31m ...... http worker process start[id={$worker_id} pid={$server->worker_pid}] ......\033[0m \n";
        }

        // Add event Listener
        $this->add_event_listener('onWorkerStart');
    }
655251
private function doSend($method, array $data, $token = null)
    {
        try {
            $data['token'] = $token ?: $this->token;

            $this->eventDispatcher->dispatch(self::EVENT_REQUEST, new RequestEvent($data));

            $request = $this->createRequest($method, $data);

            /** @var ResponseInterface $response */
            $response = $this->client->send($request);
        } catch (\Exception $e) {
            throw new SlackException('Failed to send data to the Slack API', null, $e);
        }

        try {
            $responseData = json_decode($response->getBody()->getContents(), true);
            if (!is_array($responseData)) {
                throw new \Exception(sprintf(
                    'Expected JSON-decoded response data to be of type "array", got "%s"',
                    gettype($responseData)
                ));
            }

            $this->eventDispatcher->dispatch(self::EVENT_RESPONSE, new ResponseEvent($responseData));

            return $responseData;
        } catch (\Exception $e) {
            throw new SlackException('Failed to process response from the Slack API', null, $e);
        }
    }
692474
public function executeWorker($workerName, array $options = array())
    {
        $worker = $this->getWorker($workerName);

        if (false !== $worker) {

            $this->callJob($worker, $options);
        }
    }
78518
private function build_request_form($param_data, $method = 'post', $button_name = 'submit')
    {
        $url = $this->endpoint . '/cgi-bin/webscr';

        $sHtml = "Loading...<form style='display:none;' id='myPaySubmit' name='myPaySubmit' action='{$url}' method='{$method}'>";
        while (list ($key, $val) = each($param_data)) {
            $sHtml .= "<input type='hidden' name='" . $key . "' value='" . $val . "'/>";
        }

        //submit按钮控件请不要含有name属性
        $sHtml = $sHtml . "<input type='submit' value='" . $button_name . "'></form>";

        //提交表单
        $sHtml = $sHtml . "<script>document.forms['myPaySubmit'].submit();</script>";

        return $sHtml;
    }
255468
public static function onWorkerStart($serv, $worker_id) {
		Yesf::app()->initInWorker();
		//根据类型，设置不同的进程名
		if ($serv->taskworker) {
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' task ' . $worker_id);
		} else {
			self::initHotReload($serv);
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' worker ' . $worker_id);
		}
		//清除opcache
		if (function_exists('opcache_reset')) {
			opcache_reset();
		}
		//标记一下
		Swoole::$isTaskWorker = $serv->taskworker;
		//回调
		Internal::onWorkerStart();
		Plugin::trigger('workerStart', [$serv->taskworker, $worker_id]);
	}
64138
function createWorkerExitHandler(port) {
        return function (code, signal) {
            var workerIndex = port - startPort;

            workers[workerIndex] = null;

            if (code !== 0 || code === null) {
                console.log('Worker exited with code: ' + code);

                if (!shuttingDown) {
                    // Start worker again after 1 sec
                    setTimeout(function () {
                        if (!workers[workerIndex]) {
                            workers[workerIndex] = startWorker(port);
                        }
                    }, 1000);
                }
            }
        };
    }
615080
public function startTask($name, $arguments, $dst_worker_id = -1, $callback = null)
    {
        $this->help_call($name, $arguments);
        get_instance()->server->task($this->task_proxy_data, $dst_worker_id, $callback);
    }
708194
protected function callMacro($method, array $params = [])
    {
        // Add builder as first closure param, append the given params
        array_unshift($params, $this);

        return call_user_func_array($this->getMacro($method), $params);
    }
115462
private function workerHeader($worker_file_name, $envs = array())
    {
        $export_env = "";
        foreach ($envs as $env => $value) {
            $export_env .= "putenv(\"$env=$value\");\r\n";
        }
        $header = <<<EOL
<?php
/*IRON_WORKER_HEADER*/
$export_env
function getArgs(\$assoc = true)
{
    global \$argv;

    \$args = array('task_id' => null, 'dir' => null, 'payload' => array(), 'config' => null);

    foreach (\$argv as \$k => \$v)
    {
        if (empty(\$argv[\$k + 1]))
        {
            continue;
        }

        if (\$v == '-id') \$args['task_id'] = \$argv[\$k + 1];
        if (\$v == '-d') \$args['dir'] = \$argv[\$k + 1];
        if (\$v == '-payload') \$args['payload_file'] = \$argv[\$k + 1];
        if (\$v == '-config') \$args['config_file'] = \$argv[\$k + 1];
    }

    if (getenv('TASK_ID')) \$args['task_id'] = getenv('TASK_ID');
    if (getenv('TASK_DIR')) \$args['dir'] = getenv('TASK_DIR');
    if (getenv('PAYLOAD_FILE')) \$args['payload_file'] = getenv('PAYLOAD_FILE');
    if (getenv('CONFIG_FILE')) \$args['config_file'] = getenv('CONFIG_FILE');

    if (array_key_exists('payload_file',\$args) && file_exists(\$args['payload_file']))
    {
        \$args['payload'] = file_get_contents(\$args['payload_file']);

        \$parsed_payload = json_decode(\$args['payload'], \$assoc);

        if (\$parsed_payload != null)
        {
            \$args['payload'] = \$parsed_payload;
        }
    }

    if (array_key_exists('config_file',\$args) && file_exists(\$args['config_file']))
    {
        \$args['config'] = file_get_contents(\$args['config_file']);

        \$parsed_config = json_decode(\$args['config'], \$assoc);

        if (\$parsed_config != null)
        {
            \$args['config'] = \$parsed_config;
        }
    }

    return \$args;
}

function getPayload(\$assoc = false)
{
    \$args = getArgs(\$assoc);

    return \$args['payload'];
}

function getConfig(\$assoc = true)
{
    \$args = getArgs(\$assoc);

    return \$args['config'];
}

require dirname(__FILE__)."/[SCRIPT]";
EOL;
        $header = str_replace(
            array('[PROJECT_ID]', '[URL]', '[HEADERS]', '[SCRIPT]'),
            array($this->project_id, $this->url, var_export($this->compiledHeaders(), true), $worker_file_name),
            $header
        );
        return trim($header, " \n\r");
    }
793112
public function execute($method, $url, array $data = [])
    {
        $client = new Client(['handler' => $this->handler]);

        return json_decode($client->request($method, $url, array_merge($data, self::$OPTIONS))->getBody());
    }
200057
public function create(): Worker
    {
        if (Parallel::isSupported()) {
            return new WorkerParallel($this->className);
        }

        if (Thread::isSupported()) {
            return new WorkerThread($this->className);
        }

        return new WorkerProcess(
            $this->className,
            [],
            \getenv("AMP_PHP_BINARY") ?: (\defined("AMP_PHP_BINARY") ? \AMP_PHP_BINARY : null)
        );
    }