README

Laravel 对库 https://gitlab.com/GDXbsv/pservicebus 的集成

Packagist: https://packagist.org.cn/packages/gdx/p-service-bus-laravel-package

安装

composer require gdx/p-service-bus-laravel-package

配置

在配置中添加您的传输（这些只是示例，您可以根据需要创建任意数量的传输，并赋予任意名称。示例中有 2 个传输：用于外部通信的 SNS 和用于内部通信的 SQS）

    'transports' => [
        'external' => SnsSqsTransport::class,
        'main' => in_array(env('APP_ENV'), ['production', 'staging',]) ? 'app.service_bus.transport.main' : InMemoryTransport::class,
    ],

在您的应用程序中定义这些传输

<?php declare(strict_types=1);

namespace App\Providers;

use GDXbsv\PServiceBus\Transport\Sns\SnsSqsTransport;
use GDXbsv\PServiceBus\Transport\Sqs\SqsTransport;
use Illuminate\Contracts\Foundation\Application;
use Illuminate\Foundation\Console\ClosureCommand;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Support\ServiceProvider;

class PServiceBusProvider extends ServiceProvider
{
    public function register()
    {
        $snsDsn = env('SNS_DSN', 'sns+http://key:secret@aws/000000000000?region=eu-central-1&topic=service_bus');
        $sqsDsn = env('SQS_DSN', 'sqs+http://key:secret@aws/000000000000?region=eu-central-1');
        $this->app->singleton(SnsSqsTransport::class,
            fn(Application $app) => SnsSqsTransport::ofDsn(
                $snsDsn,
                SqsTransport::ofDsn($sqsDsn . '&retries=3&queue=editing_ext'),
                $app->make('p-service-bus.build.storage')->messageNameMapIn,
            )
        );
        $this->app->singleton('app.service_bus.transport.main', fn() => SqsTransport::ofDsn($sqsDsn . '&retries=3&queue=editing_main'));
    }
}

Saga Eloquent

创建 saga

主要区别在于您需要扩展 GDXbsv\PServiceBusLaravel\Saga\SagaEloquent 并实现 getEloquentModelClass

<?php declare(strict_types=1);

namespace App\Saga;

use Doctrine\ORM\Mapping as ORM;
use GDXbsv\PServiceBus\Bus\Handling\Handle;
use GDXbsv\PServiceBus\Id;
use GDXbsv\PServiceBus\Message\TimeSpan;
use GDXbsv\PServiceBus\Saga\MessageSagaContext;
use GDXbsv\PServiceBus\Saga\SagaContext;
use GDXbsv\PServiceBus\Saga\SagaCreateMapper;
use GDXbsv\PServiceBus\Saga\SagaPropertyMapper;
use GDXbsv\PServiceBusLaravel\Saga\SagaEloquent;


/**
 * @final
 */
final class TestSaga extends SagaEloquent
{
    private Id $id;
    public string $string;
    public ?string $value = null;


    public static function getEloquentModelClass(): string
    {
        return TestSagaModel::class;
    }

    /**
     * @param Id<static> $id
     */
    private function __construct(Id $id, string $string)
    {
        $this->id = $id;
        $this->string = $string;
    }

    public static function configureHowToCreateSaga(SagaCreateMapper $mapper): void
    {
        $mapper
            ->toMessage(
                // do not forget to create handling function in a case if saga exists and to let saga know that we wait this message
                function (TestSagaCreateCommand $command, MessageSagaContext $context) {
                    return new self(new Id($command->id), $command->string);
                }
            );
    }

    public static function configureHowToFindSaga(SagaPropertyMapper $mapper): void
    {
        $mapper
            ->mapSaga(new \ReflectionProperty(TestSaga::class, 'id'))
            ->toMessage(
                function (TestSagaCommand $command, MessageSagaContext $context) {
                    return new Id($command->id);
                }
            );
    }

    /** We have to tell saga we wait this message */
    #[Handle('main', 3)]
    public function testSagaCreateCommand(TestSagaCreateCommand $command, SagaContext $context)
    {
        $this->string = $command->string;
        $context->publish(new TestsSagaOutputEvent('testHandlerFunction'));
    }
}

然后创建提供的 EloquentModel

<?php declare(strict_types=1);

namespace App\Saga;

use Illuminate\Database\Eloquent\Model;

class TestSagaModel extends Model
{
    protected $table = 'saga';
    protected $primaryKey = 'id';
    public $incrementing = false;
    protected $guarded = [];
}

之后就可以像平常一样使用了。

laravel 的命令

p-service-bus:cache:clear 清理缓存并再次搜索属性。仅在 debug=false 模式下需要。

p-service-bus:saga:eloquent:outbox:recover-messages 如果出现问题，则运行它从出站发送消息。定期运行。

p-service-bus:saga:eloquent:only-once:clean {days=30} 清理仅一次控制的老旧消息。每天运行一次。