add entity parse constraint
This commit is contained in:
@@ -0,0 +1,211 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Entity\Parse;
|
||||
|
||||
use App\Model\Model as Entity;
|
||||
use App\Model\Company;
|
||||
use App\Model\Platform;
|
||||
use App\Model\Store;
|
||||
use App\Entity\Parse\Traits\EntityParseHelper;
|
||||
use Hyperf\Amqp\Message\ConsumerMessageInterface;
|
||||
use InvalidArgumentException;
|
||||
|
||||
/**
|
||||
* EntityParse 抽象类
|
||||
*
|
||||
* 使用工厂方法模式 + 延迟初始化
|
||||
* 提供消息解析的通用框架和默认实现
|
||||
*
|
||||
* @method static static create(ConsumerMessageInterface $message)
|
||||
*/
|
||||
abstract class EntityParse implements EntityParseInterface
|
||||
{
|
||||
use EntityParseHelper;
|
||||
|
||||
protected ConsumerMessageInterface $message;
|
||||
protected ?Platform $platform = null;
|
||||
protected ?Company $company = null;
|
||||
protected ?Store $store = null;
|
||||
|
||||
/**
|
||||
* 禁止直接使用构造函数
|
||||
* 使用 create() 工厂方法创建实例
|
||||
*/
|
||||
protected function __construct()
|
||||
{
|
||||
// 空构造函数，延迟初始化
|
||||
}
|
||||
|
||||
/**
|
||||
* 工厂方法：创建解析器实例
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return static
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public static function create(ConsumerMessageInterface $message): static
|
||||
{
|
||||
$instance = new static();
|
||||
$instance->message = $message;
|
||||
$instance->initialize();
|
||||
return $instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* 延迟初始化
|
||||
* 在 message 设置后执行作用域匹配和验证
|
||||
*
|
||||
* @return void
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function initialize(): void
|
||||
{
|
||||
$this->platform = $this->platformScopeMatch($this->message);
|
||||
$this->company = $this->companyScopeMatch($this->message);
|
||||
$this->store = $this->storeScopeMatch($this->message);
|
||||
|
||||
$this->validateScope();
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证作用域对象是否有效
|
||||
*
|
||||
* @return void
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function validateScope(): void
|
||||
{
|
||||
if (!$this->platform instanceof Platform || !isset($this->platform->id)) {
|
||||
throw new InvalidArgumentException('Platform is undefined or invalid');
|
||||
}
|
||||
|
||||
if (!$this->company instanceof Company || !isset($this->company->id)) {
|
||||
throw new InvalidArgumentException('Company is undefined or invalid');
|
||||
}
|
||||
|
||||
if (!$this->store instanceof Store || !isset($this->store->id)) {
|
||||
throw new InvalidArgumentException('Store is undefined or invalid');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 平台作用域匹配 - 提供默认实现
|
||||
*
|
||||
* 从 exchange 名称中提取平台信息
|
||||
* 子类可以覆盖此方法以实现自定义逻辑
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Platform
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public function platformScopeMatch(ConsumerMessageInterface $message): Platform
|
||||
{
|
||||
return $this->extractPlatformFromExchange($message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 实体匹配 - 提供默认实现
|
||||
*
|
||||
* 从 routing key 中提取实体类型
|
||||
* 子类可以覆盖此方法以实现自定义逻辑
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Entity
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public function entityMatch(ConsumerMessageInterface $message): Entity
|
||||
{
|
||||
return $this->extractEntityFromRoutingKey($message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 唯一标识符提取 - 提供默认实现
|
||||
*
|
||||
* 默认从消息体中提取 id 或 unique_id
|
||||
* 子类可以覆盖此方法以实现自定义逻辑
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return string|int
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public function entityUniqueIdentifierExtract(ConsumerMessageInterface $message): string|int
|
||||
{
|
||||
$data = $this->extractMessageData($message);
|
||||
return $this->extractUniqueIdentifier($data);
|
||||
}
|
||||
|
||||
/**
|
||||
* 公司作用域匹配 - 抽象方法
|
||||
*
|
||||
* 必须由子类实现，因为不同平台的公司识别逻辑不同
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Company
|
||||
*/
|
||||
abstract public function companyScopeMatch(ConsumerMessageInterface $message): Company;
|
||||
|
||||
/**
|
||||
* 店铺作用域匹配 - 抽象方法
|
||||
*
|
||||
* 必须由子类实现，因为不同平台的店铺识别逻辑不同
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Store
|
||||
*/
|
||||
abstract public function storeScopeMatch(ConsumerMessageInterface $message): Store;
|
||||
|
||||
/**
|
||||
* 实体数据映射 - 抽象方法
|
||||
*
|
||||
* 必须由子类实现，因为不同平台的数据结构不同
|
||||
*
|
||||
* @param array $data
|
||||
* @param Entity $entity
|
||||
* @return Entity
|
||||
*/
|
||||
abstract public function entityMap(array $data, Entity $entity): Entity;
|
||||
|
||||
// ==================== Getter 方法 ====================
|
||||
|
||||
/**
|
||||
* 获取消息对象
|
||||
*
|
||||
* @return ConsumerMessageInterface
|
||||
*/
|
||||
public function getMessage(): ConsumerMessageInterface
|
||||
{
|
||||
return $this->message;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取平台对象
|
||||
*
|
||||
* @return Platform
|
||||
*/
|
||||
public function getPlatform(): Platform
|
||||
{
|
||||
return $this->platform;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取公司对象
|
||||
*
|
||||
* @return Company
|
||||
*/
|
||||
public function getCompany(): Company
|
||||
{
|
||||
return $this->company;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取店铺对象
|
||||
*
|
||||
* @return Store
|
||||
*/
|
||||
public function getStore(): Store
|
||||
{
|
||||
return $this->store;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,336 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Entity\Parse;
|
||||
|
||||
use Hyperf\Amqp\Message\ConsumerMessageInterface;
|
||||
use Hyperf\Contract\ConfigInterface;
|
||||
use Hyperf\Stringable\Str;
|
||||
use InvalidArgumentException;
|
||||
use Psr\Container\ContainerInterface;
|
||||
|
||||
/**
|
||||
* EntityParseFactory 工厂类
|
||||
*
|
||||
* 自动根据消息特征创建对应平台的 EntityParse 实例
|
||||
* 使用静态方法调用，无需依赖注入
|
||||
*
|
||||
* @example
|
||||
* // 基础用法
|
||||
* use App\Entity\Parse\EntityParseFactory;
|
||||
*
|
||||
* $parser = EntityParseFactory::createFromMessage($message);
|
||||
* $entity = $parser->entityMatch($message);
|
||||
* $rawData = json_decode($message->getBody(), true);
|
||||
* $entity = $parser->entityMap($rawData, $entity);
|
||||
* $entity->save();
|
||||
*
|
||||
* // 指定实体类型（覆盖自动提取）
|
||||
* $parser = EntityParseFactory::createFromMessage($message, 'order');
|
||||
*
|
||||
* 命名约定（推荐方式）：
|
||||
* - 创建符合命名约定的 Parser 类，工厂自动查找
|
||||
* - 类名：{Platform}{Entity}Parser
|
||||
* - 命名空间：App\Platform\{Platform}
|
||||
* - 示例：App\Platform\Tmall\TmallOrderParser
|
||||
*
|
||||
* 查找顺序：
|
||||
* 1. App\Platform\Tmall\TmallOrderParser
|
||||
* 2. App\Platform\Tmall\OrderParser
|
||||
* 3. App\Platform\TmallOrderParser
|
||||
*
|
||||
* 如果找不到对应的 Parser 类，将抛出 InvalidArgumentException
|
||||
*
|
||||
* 配置文件（可选）：
|
||||
* - 如需覆盖默认行为，可在 config/autoload/entity_parse.php 中配置
|
||||
* - 支持通配符 '*' 匹配所有实体类型
|
||||
*/
|
||||
class EntityParseFactory
|
||||
{
|
||||
/**
|
||||
* 平台与 Parser 类的映射表
|
||||
* 格式：['platform_name' => ['entity_type' => ParserClass]]
|
||||
*
|
||||
* @var array<string, array<string, class-string<EntityParse>>>
|
||||
*/
|
||||
private static array $registry = [];
|
||||
|
||||
/**
|
||||
* 容器实例
|
||||
*/
|
||||
private static ?ContainerInterface $container = null;
|
||||
|
||||
/**
|
||||
* 配置实例
|
||||
*/
|
||||
private static ?ConfigInterface $config = null;
|
||||
|
||||
/**
|
||||
* 设置容器（由框架注入）
|
||||
*
|
||||
* @param ContainerInterface $container
|
||||
* @return void
|
||||
*/
|
||||
public static function setContainer(ContainerInterface $container): void
|
||||
{
|
||||
self::$container = $container;
|
||||
self::$config = $container->get(ConfigInterface::class);
|
||||
self::loadConfig();
|
||||
}
|
||||
|
||||
/**
|
||||
* 从配置文件加载平台映射
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
private static function loadConfig(): void
|
||||
{
|
||||
if (!self::$config) {
|
||||
return;
|
||||
}
|
||||
|
||||
$parsers = self::$config->get('entity_parse.platforms', []);
|
||||
|
||||
foreach ($parsers as $platform => $entities) {
|
||||
foreach ($entities as $entityType => $parserClass) {
|
||||
self::register($platform, $entityType, $parserClass);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据消息自动创建 Parser 实例（静态方法）
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @param string|null $entityType 实体类型（可选，如果不指定则从 routing key 中提取）
|
||||
* @return EntityParseInterface
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public static function createFromMessage(
|
||||
ConsumerMessageInterface $message,
|
||||
?string $entityType = null
|
||||
): EntityParseInterface {
|
||||
// 1. 从 exchange 中提取平台名称
|
||||
$platformName = self::extractPlatformName($message);
|
||||
|
||||
// 2. 如果未指定实体类型，从 routing key 中提取
|
||||
if ($entityType === null) {
|
||||
$entityType = self::extractEntityType($message);
|
||||
}
|
||||
|
||||
// 3. 获取对应的 Parser 类
|
||||
$parserClass = self::resolveParserClass($platformName, $entityType);
|
||||
|
||||
// 4. 创建并返回 Parser 实例
|
||||
return $parserClass::create($message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 exchange 中提取平台名称
|
||||
*
|
||||
* 规则：exchange 格式为 "platform.exchange"
|
||||
* 例如：tmall.exchange -> tmall
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return string
|
||||
*/
|
||||
private static function extractPlatformName(ConsumerMessageInterface $message): string
|
||||
{
|
||||
$exchange = $message->getExchange();
|
||||
|
||||
$platformName = Str::of($exchange)
|
||||
->before('.')
|
||||
->lower()
|
||||
->toString();
|
||||
|
||||
if (empty($platformName)) {
|
||||
throw new InvalidArgumentException("Cannot extract platform name from exchange: {$exchange}");
|
||||
}
|
||||
|
||||
return $platformName;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 routing key 中提取实体类型
|
||||
*
|
||||
* 规则：routing key 格式为 "entity.action"
|
||||
* 例如：order.create -> order
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return string
|
||||
*/
|
||||
private static function extractEntityType(ConsumerMessageInterface $message): string
|
||||
{
|
||||
$routingKey = $message->getRoutingKey();
|
||||
|
||||
$entityType = Str::of($routingKey)
|
||||
->before('.')
|
||||
->lower()
|
||||
->toString();
|
||||
|
||||
if (empty($entityType)) {
|
||||
throw new InvalidArgumentException("Cannot extract entity type from routing key: {$routingKey}");
|
||||
}
|
||||
|
||||
return $entityType;
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析 Parser 类名
|
||||
*
|
||||
* 优先级：
|
||||
* 1. 手动注册的映射
|
||||
* 2. 配置文件中的映射
|
||||
* 3. 命名约定自动查找
|
||||
*
|
||||
* @param string $platformName
|
||||
* @param string $entityType
|
||||
* @return class-string<EntityParse>
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
private static function resolveParserClass(string $platformName, string $entityType): string
|
||||
{
|
||||
// 1. 检查手动注册的映射
|
||||
if (isset(self::$registry[$platformName][$entityType])) {
|
||||
return self::$registry[$platformName][$entityType];
|
||||
}
|
||||
|
||||
// 2. 检查通配符映射（支持 * 匹配所有实体类型）
|
||||
if (isset(self::$registry[$platformName]['*'])) {
|
||||
return self::$registry[$platformName]['*'];
|
||||
}
|
||||
|
||||
// 3. 使用命名约定自动查找
|
||||
$parserClass = self::findByConvention($platformName, $entityType);
|
||||
|
||||
if ($parserClass) {
|
||||
// 自动注册以提高后续性能
|
||||
self::register($platformName, $entityType, $parserClass);
|
||||
return $parserClass;
|
||||
}
|
||||
|
||||
throw new InvalidArgumentException(
|
||||
"Cannot find parser for platform '{$platformName}' and entity '{$entityType}'. " .
|
||||
"Please register it using EntityParseFactory::register() or create a parser class following the naming convention."
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据命名约定查找 Parser 类
|
||||
*
|
||||
* 命名约定：
|
||||
* 1. 类名：{Platform}{Entity}Parser
|
||||
* 2. 命名空间：App\Platform\{Platform}
|
||||
*
|
||||
* 示例：
|
||||
* - Platform: tmall, Entity: order -> App\Platform\Tmall\TmallOrderParser
|
||||
* - Platform: shopee, Entity: product -> App\Platform\Shopee\ShopeeProductParser
|
||||
*
|
||||
* @param string $platformName
|
||||
* @param string $entityType
|
||||
* @return class-string<EntityParse>|null
|
||||
*/
|
||||
private static function findByConvention(string $platformName, string $entityType): ?string
|
||||
{
|
||||
// 构建类名
|
||||
$platformPascal = Str::of($platformName)->ucfirst()->toString();
|
||||
$entityPascal = Str::of($entityType)->ucfirst()->toString();
|
||||
|
||||
// 尝试的类名列表
|
||||
$possibleClasses = [
|
||||
// App\Platform\Tmall\TmallOrderParser
|
||||
"App\\Platform\\{$platformPascal}\\{$platformPascal}{$entityPascal}Parser",
|
||||
// App\Platform\Tmall\OrderParser
|
||||
"App\\Platform\\{$platformPascal}\\{$entityPascal}Parser",
|
||||
// App\Platform\TmallOrderParser
|
||||
"App\\Platform\\{$platformPascal}{$entityPascal}Parser",
|
||||
];
|
||||
|
||||
foreach ($possibleClasses as $class) {
|
||||
if (class_exists($class)) {
|
||||
// 验证类是否继承自 EntityParse
|
||||
if (is_subclass_of($class, EntityParse::class)) {
|
||||
return $class;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 手动注册平台与 Parser 的映射关系
|
||||
*
|
||||
* @param string $platformName 平台名称（小写）
|
||||
* @param string $entityType 实体类型（小写），使用 '*' 表示匹配所有实体类型
|
||||
* @param class-string<EntityParse> $parserClass Parser 类名
|
||||
* @return void
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
public static function register(string $platformName, string $entityType, string $parserClass): void
|
||||
{
|
||||
// 验证 Parser 类是否存在
|
||||
if (!class_exists($parserClass)) {
|
||||
throw new InvalidArgumentException("Parser class '{$parserClass}' does not exist");
|
||||
}
|
||||
|
||||
// 验证 Parser 类是否继承自 EntityParse
|
||||
if (!is_subclass_of($parserClass, EntityParse::class)) {
|
||||
throw new InvalidArgumentException(
|
||||
"Parser class '{$parserClass}' must extend " . EntityParse::class
|
||||
);
|
||||
}
|
||||
|
||||
// 注册映射
|
||||
self::$registry[$platformName][$entityType] = $parserClass;
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量注册平台映射
|
||||
*
|
||||
* @param array<string, array<string, class-string<EntityParse>>> $mappings
|
||||
* @return void
|
||||
*/
|
||||
public static function registerBatch(array $mappings): void
|
||||
{
|
||||
foreach ($mappings as $platform => $entities) {
|
||||
foreach ($entities as $entityType => $parserClass) {
|
||||
self::register($platform, $entityType, $parserClass);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取所有已注册的映射
|
||||
*
|
||||
* @return array<string, array<string, class-string<EntityParse>>>
|
||||
*/
|
||||
public static function getRegistry(): array
|
||||
{
|
||||
return self::$registry;
|
||||
}
|
||||
|
||||
/**
|
||||
* 清空所有注册的映射（主要用于测试）
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
public static function clearRegistry(): void
|
||||
{
|
||||
self::$registry = [];
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查指定平台和实体类型的 Parser 是否已注册
|
||||
*
|
||||
* @param string $platformName
|
||||
* @param string $entityType
|
||||
* @return bool
|
||||
*/
|
||||
public static function hasParser(string $platformName, string $entityType): bool
|
||||
{
|
||||
return isset(self::$registry[$platformName][$entityType])
|
||||
|| isset(self::$registry[$platformName]['*']);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Entity\Parse;
|
||||
|
||||
use Hyperf\Amqp\Message\ConsumerMessageInterface;
|
||||
use App\Model\Company;
|
||||
use App\Model\Platform;
|
||||
use App\Model\Store;
|
||||
use App\Model\Model as Entity;
|
||||
|
||||
/**
|
||||
* EntityParseInterface 接口
|
||||
*
|
||||
* 定义消息解析器的标准接口
|
||||
*/
|
||||
interface EntityParseInterface
|
||||
{
|
||||
/**
|
||||
* 公司作用域匹配
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Company
|
||||
*/
|
||||
public function companyScopeMatch(ConsumerMessageInterface $message): Company;
|
||||
|
||||
/**
|
||||
* 平台作用域匹配
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Platform
|
||||
*/
|
||||
public function platformScopeMatch(ConsumerMessageInterface $message): Platform;
|
||||
|
||||
/**
|
||||
* 店铺作用域匹配
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Store
|
||||
*/
|
||||
public function storeScopeMatch(ConsumerMessageInterface $message): Store;
|
||||
|
||||
/**
|
||||
* 实体类型匹配
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Entity
|
||||
*/
|
||||
public function entityMatch(ConsumerMessageInterface $message): Entity;
|
||||
|
||||
/**
|
||||
* 实体数据映射
|
||||
*
|
||||
* @param array $data
|
||||
* @param Entity $entity
|
||||
* @return Entity
|
||||
*/
|
||||
public function entityMap(array $data, Entity $entity): Entity;
|
||||
|
||||
/**
|
||||
* 提取实体唯一标识符
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return string|int
|
||||
*/
|
||||
public function entityUniqueIdentifierExtract(ConsumerMessageInterface $message): string|int;
|
||||
|
||||
/**
|
||||
* 获取消息对象
|
||||
*
|
||||
* @return ConsumerMessageInterface
|
||||
*/
|
||||
public function getMessage(): ConsumerMessageInterface;
|
||||
|
||||
/**
|
||||
* 获取平台对象
|
||||
*
|
||||
* @return Platform
|
||||
*/
|
||||
public function getPlatform(): Platform;
|
||||
|
||||
/**
|
||||
* 获取公司对象
|
||||
*
|
||||
* @return Company
|
||||
*/
|
||||
public function getCompany(): Company;
|
||||
|
||||
/**
|
||||
* 获取店铺对象
|
||||
*
|
||||
* @return Store
|
||||
*/
|
||||
public function getStore(): Store;
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Entity\Parse\Traits;
|
||||
|
||||
use App\Model\Model as Entity;
|
||||
use App\Model\Platform;
|
||||
use Hyperf\Amqp\Message\ConsumerMessageInterface;
|
||||
use InvalidArgumentException;
|
||||
use Hyperf\Stringable\Str;
|
||||
|
||||
/**
|
||||
* EntityParseHelper Trait
|
||||
*
|
||||
* 提供通用的消息解析辅助方法
|
||||
*/
|
||||
trait EntityParseHelper
|
||||
{
|
||||
/**
|
||||
* 从 exchange 名称中提取平台信息
|
||||
*
|
||||
* 规则：exchange 格式为 "platform.exchange"
|
||||
* 例如：tmall.exchange -> Tmall 平台
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Platform
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function extractPlatformFromExchange(ConsumerMessageInterface $message): Platform
|
||||
{
|
||||
$platformName = Str::of($message->getExchange())
|
||||
->before('.')
|
||||
->ucfirst()
|
||||
->toString();
|
||||
|
||||
$platform = Platform::where('name', '=', $platformName)->first();
|
||||
|
||||
if (!$platform) {
|
||||
throw new InvalidArgumentException("Platform name '{$platformName}' does not exist!");
|
||||
}
|
||||
|
||||
return $platform;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 routing key 中提取实体类型
|
||||
*
|
||||
* 规则：routing key 格式为 "entity.platform"
|
||||
* 例如：order.tmall -> Order 实体
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return Entity
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function extractEntityFromRoutingKey(ConsumerMessageInterface $message): Entity
|
||||
{
|
||||
$entityName = Str::of($message->getRoutingKey())
|
||||
->before('.')
|
||||
->ucfirst()
|
||||
->toString();
|
||||
|
||||
$entityClass = "App\\Model\\{$entityName}";
|
||||
|
||||
if (!class_exists($entityClass)) {
|
||||
throw new InvalidArgumentException("Entity class '{$entityClass}' does not exist");
|
||||
}
|
||||
|
||||
$entity = new $entityClass();
|
||||
|
||||
if (!$entity instanceof Entity) {
|
||||
throw new InvalidArgumentException("Class '{$entityClass}' is not an instance of Entity");
|
||||
}
|
||||
|
||||
return $entity;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从消息体中提取 JSON 数据
|
||||
*
|
||||
* @param ConsumerMessageInterface $message
|
||||
* @return array
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function extractMessageData(ConsumerMessageInterface $message): array
|
||||
{
|
||||
$body = $message->getBody();
|
||||
|
||||
if (empty($body)) {
|
||||
throw new InvalidArgumentException('Message body is empty');
|
||||
}
|
||||
|
||||
$data = json_decode($body, true);
|
||||
|
||||
if (json_last_error() !== JSON_ERROR_NONE) {
|
||||
throw new InvalidArgumentException('Invalid JSON in message body: ' . json_last_error_msg());
|
||||
}
|
||||
|
||||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从消息数据中提取唯一标识符
|
||||
*
|
||||
* 优先级：id > unique_id > 其他自定义字段
|
||||
*
|
||||
* @param array $data
|
||||
* @param string|null $customField 自定义标识符字段名
|
||||
* @return string|int
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function extractUniqueIdentifier(array $data, ?string $customField = null): string|int
|
||||
{
|
||||
// 优先使用自定义字段
|
||||
if ($customField !== null && isset($data[$customField])) {
|
||||
return $data[$customField];
|
||||
}
|
||||
|
||||
// 默认查找 id
|
||||
if (isset($data['id'])) {
|
||||
return $data['id'];
|
||||
}
|
||||
|
||||
// 查找 unique_id
|
||||
if (isset($data['unique_id'])) {
|
||||
return $data['unique_id'];
|
||||
}
|
||||
|
||||
throw new InvalidArgumentException('Cannot extract unique identifier from message data');
|
||||
}
|
||||
|
||||
/**
|
||||
* 从消息数据中安全地提取字段值
|
||||
*
|
||||
* @param array $data
|
||||
* @param string $field
|
||||
* @param mixed $default
|
||||
* @return mixed
|
||||
*/
|
||||
protected function getDataField(array $data, string $field, mixed $default = null): mixed
|
||||
{
|
||||
return $data[$field] ?? $default;
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证消息数据中的必需字段
|
||||
*
|
||||
* @param array $data
|
||||
* @param array $requiredFields
|
||||
* @return void
|
||||
* @throws InvalidArgumentException
|
||||
*/
|
||||
protected function validateRequiredFields(array $data, array $requiredFields): void
|
||||
{
|
||||
$missingFields = [];
|
||||
|
||||
foreach ($requiredFields as $field) {
|
||||
if (!isset($data[$field])) {
|
||||
$missingFields[] = $field;
|
||||
}
|
||||
}
|
||||
|
||||
if (!empty($missingFields)) {
|
||||
throw new InvalidArgumentException(
|
||||
'Missing required fields: ' . implode(', ', $missingFields)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user