datahub/backend/app/Command/TestErrorQueueCommand.php

<?php

declare(strict_types=1);

namespace App\Command;

use App\Platform\ErrorProducer;
use Hyperf\Amqp\Producer;
use Hyperf\Command\Command as HyperfCommand;
use Hyperf\Command\Annotation\Command;
use Hyperf\Context\ApplicationContext;
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Container\ContainerInterface;
use Exception;

#[Command]
class TestErrorQueueCommand extends HyperfCommand
{
    protected ?string $name = 'app:test:error-queue';

    public function __construct(protected ContainerInterface $container)
    {
        parent::__construct();
    }

    public function configure()
    {
        parent::configure();
        $this->setDescription('测试向 errors.queue 发送消息,验证 consumer 权限');
    }

    public function handle()
    {
        $producer = ApplicationContext::getContainer()->get(Producer::class);

        $testMessage = [
            'test_id' => uniqid('test_', true),
            'message' => 'This is a test error message',
            'timestamp' => date('c'),
        ];

        $this->line("======================================");
        $this->line("测试 1: 使用空 exchange (默认 exchange) 发送到 errors.queue");
        $this->line("======================================");

        try {
            $producer->produce(
                new class($testMessage) extends \Hyperf\Amqp\Message\ProducerMessage {
                    protected string $exchange = '';
                    protected string|array $routingKey = 'errors.queue';
                    protected string $poolName = 'default_consumer';
                    protected array $properties = [
                        'delivery_mode' => 2,
                    ];

                    public function __construct(array $data)
                    {
                        $this->payload = $data;
                    }
                }
            );
            $this->info("✓ 测试 1 成功: 消息已发送到 errors.queue");
        } catch (\Throwable $e) {
            $this->error("✗ 测试 1 失败: " . $e->getMessage());
        }

        $this->line("");
        $this->line("======================================");
        $this->line("测试 2: 使用通用 errors.exchange 发送到 errors.queue");
        $this->line("======================================");

        try {
            $producer->produce(
                new class($testMessage) extends \Hyperf\Amqp\Message\ProducerMessage {
                    protected string $exchange = 'errors.exchange';
                    protected string|array $routingKey = 'error.test';
                    protected string $poolName = 'default_consumer';
                    protected array $properties = [
                        'delivery_mode' => 2,
                    ];

                    public function __construct(array $data)
                    {
                        $this->payload = $data;
                    }
                }
            );
            $this->info("✓ 测试 2 成功: 消息已通过 errors.exchange 发送");
        } catch (\Throwable $e) {
            $this->error("✗ 测试 2 失败: " . $e->getMessage());
        }

        $this->line("");
        $this->line("======================================");
        $this->line("测试 3: 使用 ErrorProducer 类发送到 errors.queue");
        $this->line("======================================");

        try {
            // 模拟一个原始消息
            $originalData = [
                'message_id' => 'test_msg_123',
                'timestamp' => date('c'),
                'platform' => 'tmall',
                'data_type' => 'order',
                'meta' => [
                    'platform_id' => 2,
                    'company_id' => 1,
                    'store_id' => 10,
                ],
                'data' => ['test' => 'data'],
            ];

            $amqpMessage = new AMQPMessage(json_encode($originalData));
            $testException = new Exception('Test error for ErrorProducer');

            $errorProducer = new ErrorProducer($amqpMessage, $testException, 3);
            $producer->produce($errorProducer);

            $this->info("✓ 测试 3 成功: 使用 ErrorProducer 类发送消息");
        } catch (\Throwable $e) {
            $this->error("✗ 测试 3 失败: " . $e->getMessage());
        }

        $this->line("");
        $this->line("现在请检查 errors.queue 中是否有消息:");
        $this->line("php bin/hyperf.php app:mq:status");

        return 0;
    }
}