117 lines
2.9 KiB
PHP
117 lines
2.9 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace App\Platform;
|
|
|
|
use Exception;
|
|
use Hyperf\Amqp\Annotation\Producer;
|
|
use Hyperf\Amqp\Message\ProducerMessage;
|
|
|
|
// exchange 示例值 'tmall.exchange', routeKey 示例值 'order.tmall'
|
|
#[Producer('', '')]
|
|
class OrderProducer extends ProducerMessage
|
|
{
|
|
/**
|
|
* 指定使用的连接池名称
|
|
*/
|
|
protected string $poolName = 'default_producer';
|
|
|
|
/**
|
|
* VHost
|
|
*/
|
|
protected string $vhost = 'datahub';
|
|
|
|
/**
|
|
* 消息持久化
|
|
*/
|
|
protected array $properties = [
|
|
'delivery_mode' => 2, // 持久化消息
|
|
];
|
|
|
|
|
|
protected string $exchange = '';
|
|
protected string $entityType = 'order';
|
|
protected string|array $routingKey = '';
|
|
|
|
|
|
/**
|
|
* 构造消息
|
|
*
|
|
* @param array $data 订单数据
|
|
* @return string
|
|
*/
|
|
public function __construct(array $data = [])
|
|
{
|
|
if(empty($this->exchange)){
|
|
throw new Exception('exchange is not set!');
|
|
}
|
|
|
|
if(empty($this->routingKey)){
|
|
throw new Exception('routingKey is not set!');
|
|
}
|
|
|
|
if(!empty($this->routingKey)){
|
|
$this->entityType = \explode( '.', $this->routingKey)[0];
|
|
}
|
|
|
|
if (!empty($data)) {
|
|
$this->payload = $this->buildMessage($data);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 构建消息格式
|
|
*
|
|
* @param array $data 原始订单数据
|
|
* @return array
|
|
*/
|
|
protected function buildMessage(array $data): array
|
|
{
|
|
// 根据 RabbitMQ.md 中定义的消息格式规范
|
|
return [
|
|
'message_id' => $this->generateMessageId($data),
|
|
'timestamp' => date('c'), // ISO 8601 格式
|
|
'platform' => 'platform_unkonw',
|
|
'data_type' => 'order',
|
|
'meta' => [
|
|
'platform_id' => $data['platform_id'] ?? null,
|
|
'company_id' => $data['company_id'] ?? null,
|
|
'store_id' => $data['store_id'] ?? null,
|
|
'platform_store_id' => $data['platform_store_id'] ?? null,
|
|
'unique_id' => $data['unique_id'] ?? null,
|
|
'source_system' => 'origin_unknow',
|
|
'retry_count' => 0,
|
|
'data_version' => $data['data_version'] ?? time(),
|
|
],
|
|
'data' => $data['raw_data'],
|
|
];
|
|
}
|
|
|
|
/**
|
|
* 生成消息ID
|
|
*
|
|
* 格式: {prefix}#{app_id}#{company_id}#{platform_id}#{store_id}#{entity_type}#{entity_id}
|
|
*
|
|
* @param array $data
|
|
* @return string
|
|
*/
|
|
protected function generateMessageId(array $data): string
|
|
{
|
|
$company_id = $data['company_id'];
|
|
$platform_id = $data['platform_id'];
|
|
$store_id = $data['store_id'];
|
|
|
|
$unique_id = $data['unique_id'];
|
|
|
|
return sprintf(
|
|
'%s#%s#%s#%s#%s',
|
|
$company_id,
|
|
$platform_id,
|
|
$store_id,
|
|
$this->entityType,
|
|
$unique_id
|
|
);
|
|
}
|
|
}
|