目标平台查询适配器：实现dispatch，handleResponse，connect方法
<?php
namespace Adapter\PlatformName;
use Domain\Datahub\Instance\Storage\DataStatus;
use Domain\Datahub\Instance\Adapter\Adapter;
use Domain\Datahub\Instance\LogMessage;
use Domain\Datahub\Instance\Storage\LogStatus;
class PlatformNameExecuteAdapter extends Adapter {
const DIRECTION = 'target';
private $times = 0;
// 调度方法
public function dispatch() {
$this->times++;
if ($this->times >= 30) {
$this->asynTargetJobDispatch(10);
return;
}
$data = $this->getDataStorage()->fetch();
if (count($data) === 0) {
return $this->_returnDispatch();
}
$request = $this->generateRequestParams($data);
$request = $this->removeNull($request);
if (!$request) {
$this->getLogStorage()->insertOne(['text' => LogMessage::DISPATCH_TARGET_REQUEST_ERROR, 'request' => $request], LogStatus::ERROR);
$this->dispatch();
return;
}
$jobId = $this->getAsynTargetJobStorage()->insertOne($this->metaData['api'], [$request], $this->getDataStorage()->ids, $this->getDataStorage()->dataRange);
$this->getDataStorage()->setFetchStatus(DataStatus::QUEUE, null, null, new \MongoDB\BSON\ObjectId($jobId));
$this->jobs[] = $jobId;
$this->asynTargetJob(round($this->asynTimes), $jobId);
$this->asynTimes += 1.4;
$this->dispatch();
return true;
}
}目标平台写入 $adapter->dispatch() 步骤：
- 递增调度次数
$times++，如果超过最大调度次数，则排队等待10秒再进行下一次调度。 - 从 MongoDB 获取一批待处理数据。
- 数据和元数据生成请求参数。
- 将任务写入队列存储器并返回
jobId。 - 任务开始排队（自定义延迟）。
- 标记数据为正在排队状态。
- 递归调用，进行下一次调度。
目标平台写入 $adapter->handleResponse()：
public function handleResponse($response, $jobId = null) {
$this->getLogStorage()->insertOne(['text' => 'handleResponse', 'response' => $response], LogStatus::RECORD);
if ($response['Success'] != true) {
return $this->handleError($response, $jobId);
}
$this->getAsynTargetJobStorage()->updateResponse($jobId, DataStatus::FINISHED, $response, [], null, $this->active);
$this->handleSuccessCallback($response, $jobId);
return $response;
}错误日志方法：
public function handleError($response, $jobId = null) {
$throw = new PlatformThrowable($this);
$throw->handle($jobId, $response);
$this->getAsynTargetJobStorage()->updateResponse($jobId, DataStatus::ERROR, $response, [], null, $this->active);
$this->getLogStorage()->insertOne(['text' => LogMessage::INVOKE_FAIL, 'response' => $response], LogStatus::ERROR);
return $response;
}目标平台写入 $adapter->connect() 方法与同源平台方法一致。