README

EventSauceBundle 3.0

eventsauce 的官方 文档

支持

• Doctrine3 事件存储
• Symfony messenger 消息分发器
• 反腐败层
• 事件分发器
• 消息出盒
• 快照 Doctrine 存储库，版本控制，条件持久化
• 每个聚合类型的所有事件都在表中
• 为每个聚合生成迁移

旧版本

• 2.0

要求

• PHP >=8.2
• Symfony ^6.2

安装

composer require andreo/eventsauce-bundle

// config/bundles.php

return [
    Andreo\EventSauceBundle\AndreoEventSauceBundle::class => ['all' => true],
];

简介

以下配置展示了默认值和一些示例值。
请注意，大多数默认配置值无需配置。

时钟

andreo_event_sauce:
  clock:
    timezone: UTC

有用的别名

EventSauce\Clock\Clock: EventSauce\Clock\SystemClock

消息存储

Doctrine 3

andreo_event_sauce:
  #...
  message_storage:
    repository:
      doctrine_3:
        enabled: true
        json_encode_flags: []
        connection: doctrine.dbal.default_connection
        table_name: event_store

需要

• doctrine/dbal

消息分发器

SynchronousMessageDispatcher

andreo_event_sauce:
  #...
  message_dispatcher: # chain of message dispatchers
    foo_dispatcher:
      type:
        sync: true
    bar_dispatcher:
      type:
        sync: true

事件消费者

use EventSauce\EventSourcing\MessageConsumer;
use Andreo\EventSauceBundle\Attribute\AsSyncMessageConsumer;
use EventSauce\EventSourcing\EventConsumption\EventConsumer;
use Andreo\EventSauce\Messenger\EventConsumer\InjectedHandleMethodInflector;
use EventSauce\EventSourcing\Message;

#[AsSyncMessageConsumer(dispatcher: 'foo_dispatcher')]
final class FooBarEventConsumer extends EventConsumer 
{
    // copy-paste trait for inject HandleMethodInflector of EventSauce
    use InjectedHandleMethodInflector;
    
    public function __construct(
        private HandleMethodInflector $handleMethodInflector
    ){}
    
    public function onFooCreated(FooCreated $fooCreated, Message $message): void {
    }
    
    public function onBarCreated(BarCreated $barCreated, Message $message): void {
    }
}

手动注册同步消费者的示例
(不带属性和自动配置)

services:
  #...
  App\Consumer\FooBarEventConsumer:
    tags:
      -
        name: andreo.eventsauce.sync_message_consumer

MessengerMessageDispatcher

使用 Symfony messenger 进行分发

安装 andreo/eventsauce-messenger

composer require andreo/eventsauce-messenger

andreo_event_sauce:
 #...
  message_dispatcher: # chain of message dispatchers
    foo_dispatcher:
      type:
        messenger:
          bus: event_bus # bus alias from messenger config

它注册了处理事件酱消息中间件的别名

$busAlias.handle_eventsauce_message: Andreo\EventSauce\Messenger\Middleware\HandleEventSauceMessageMiddleware

根据上述配置更新 messenger 配置

framework:
  messenger:
    #...
    buses:
      event_bus:
        default_middleware: false # disable default middleware order
        middleware:
          - 'add_bus_name_stamp_middleware': ['event_bus']
          - 'dispatch_after_current_bus'
          - 'failed_message_processing_middleware'
          - 'send_message'
          - 'event_bus.handle_eventsauce_message' # our middleware should be placed after send_message and before default handle massage middleware (if you use)
          - 'handle_message'

事件消费者

use Andreo\EventSauce\Messenger\EventConsumer\InjectedHandleMethodInflector;
use EventSauce\EventSourcing\EventConsumption\EventConsumer;
use EventSauce\EventSourcing\EventConsumption\HandleMethodInflector;
use Andreo\EventSauce\Messenger\Attribute\AsEventSauceMessageHandler;
use EventSauce\EventSourcing\Message;

final class FooBarEventConsumer extends EventConsumer
{
    use InjectedHandleMethodInflector;

    public function __construct(
        private HandleMethodInflector $handleMethodInflector
    )
    {}

    #[AsEventSauceMessageHandler(bus: 'fooBus')]
    public function onFooCreated(FooCreated $fooCreated, Message $message): void
    {
    }

    #[AsEventSauceMessageHandler(bus: 'barBus')]
    public function onBarCreated(BarCreated $barCreated, Message $message): void
    {
    }
}

有用的别名

EventSauce\EventSourcing\EventConsumption\HandleMethodInflector: EventSauce\EventSourcing\EventConsumption\InflectHandlerMethodsFromType

消息分发器标签 (如果需要手动注册分发器)

andreo.eventsauce.message_dispatcher

反腐败层

andreo_event_sauce:
  #...
  acl: true

为 消息分发器 启用 (通过配置)

andreo_event_sauce:
 #...
  message_dispatcher:
    fooDispatcher:
      type:
        messenger:
          bus: fooBus
      acl:
        enabled: true
        message_filter_strategy:
          before_translate: match_all # or match_any
          after_translate: match_all # or match_any

为 消息消费者 启用

use Andreo\EventSauceBundle\Attribute\EnableAcl;
use Andreo\EventSauceBundle\Enum\MessageFilterStrategy;
use EventSauce\EventSourcing\EventConsumption\EventConsumer;

#[EnableAcl]
final class FooHandler extends EventConsumer
{
    #[AsEventSauceMessageHandler(
        handles: FooEvent::class // If you use a translator, "handles" must be configured.
    )]
    public function onFooCreated(BarEvent $barEvent): void
    {
        // ...
    }
}

手动注册 acl 消费者（或分发器）的示例
(不带属性和自动配置)

services:
  #...
  App\Consumer\FooConsumer:
    tags:
      -
        name: andreo.eventsauce.acl
        message_filter_strategy_before_translate: match_all # or match_any
        message_filter_strategy_after_translate: match_all # or match_any

消息翻译器

use EventSauce\EventSourcing\AntiCorruptionLayer\MessageTranslator;
use Andreo\EventSauceBundle\Attribute\AsMessageTranslator;
use EventSauce\EventSourcing\Message;

#[AsMessageTranslator] 
final readonly class FooMessageTranslator implements MessageTranslator
{
    public function translateMessage(Message $message): Message
    {
        assert($message->payload() instanceof FooEvent);
        // ...
           
        return new Message(new BarEvent());
    }
}

手动注册消息过滤器的示例
(不带属性和自动配置)

services:
  #...
  App\Acl\FooMessageTranslator:
    tags:
      -
        name: andreo.eventsauce.acl.message_translator
        priority: 0
        owners: []

消息过滤器

消息过滤器策略

match_all - 所有过滤器通过条件
match_any - 任何过滤器通过条件

消息过滤器

use EventSauce\EventSourcing\AntiCorruptionLayer\MessageFilter;
use Andreo\EventSauceBundle\Attribute\AsMessageFilter;
use Andreo\EventSauceBundle\Enum\MessageFilterTrigger;

#[AsMessageFilter(MessageFilterTrigger::BEFORE_TRANSLATE)] // or after AFTER_TRANSLATE
final readonly class FooMessageFilter implements MessageFilter
{
    public function allows(Message $message): bool
    {
    }
}

手动注册消息过滤器的示例
(不带属性和自动配置)

services:
  #...
  App\Acl\FooMessageFilter:
    tags:
      -
        name: andreo.eventsauce.acl.message_filter
        trigger: before_translate # or after_translate
        priority: 0
        owners: []

消息翻译器或过滤器的所有者

例如，我们使用 Translator，但 Filter 的工作方式相同

use EventSauce\EventSourcing\AntiCorruptionLayer\MessageTranslator;
use Andreo\EventSauceBundle\Attribute\AsMessageTranslator;
use EventSauce\EventSourcing\MessageConsumer;
use EventSauce\EventSourcing\MessageDispatcher;

// Translator will be using through all dispatchers as MessageDispatcher::class (or consumers as MessageConsumer::class)
// Single FooConsumer (or single FooDispatcher) uses translator also
#[AsMessageTranslator(owners: [MessageDispatcher::class, FooConsumer::class])]
final readonly class FooMessageTranslator implements MessageTranslator
{
    public function translateMessage(Message $message): Message
    {
    }
}

事件分发器

andreo_event_sauce:
  # ...
  event_dispatcher:
    enabled: false
    message_outbox:
      enabled: false
      table_name: event_message_outbox # it will be used if the outbox config has doctrine repository
      relay_id: event_dispatcher_relay # it is used by consume outbox message command

事件分发器的示例

use EventSauce\EventSourcing\EventDispatcher;

final readonly class FooHandler
{
    public function __construct(
        private EventDispatcher $eventDispatcher
    ) {
    }

    public function handle(): void
    {
        $this->eventDispatcher->dispatch(
            new FooEvent()
        );
    }
}

升级器

andreo_event_sauce:
    #...
  upcaster:
    enabled: false
    trigger: before_unserialize # or after_unserialize (on payload or on object of message)

反序列化之前

use Andreo\EventSauceBundle\Attribute\AsUpcaster;
use EventSauce\EventSourcing\Upcasting\Upcaster;

#[AsUpcaster(aggregateClass: FooAggregate::class, version: 2)]
final class FooEventV2Upcaster implements Upcaster {

    public function upcast(array $message): array
    {
    }
}

反序列化之后

安装 andreo/eventsauce-upcasting

composer require andreo/eventsauce-upcasting

use EventSauce\EventSourcing\Message;
use Andreo\EventSauce\Upcasting\MessageUpcaster\MessageUpcaster;
use Andreo\EventSauce\Upcasting\MessageUpcaster\Event;
use Andreo\EventSauceBundle\Attribute\AsUpcaster;

#[AsUpcaster(aggregateClass: FooAggregate::class, version: 2)]
final class SomeEventV2Upcaster implements MessageUpcaster {

    #[Event(event: FooEvent::class)]
    public function upcast(Message $message): Message
    {
    }
}

手动注册（不带属性和自动配置）的示例

services:
  #...
  App\Upcaster\FooUpcaster:
    tags:
      -
        name: andreo.eventsauce.upcaster
        class: App\Domain\FooAggregate
        version: 2

消息装饰器

andreo_event_sauce:
    #...
    message_decorator: true

use Andreo\EventSauceBundle\Attribute\AsMessageDecorator;
use EventSauce\EventSourcing\Message;
use EventSauce\EventSourcing\MessageDecorator;

#[AsMessageDecorator]
final class FooDecorator implements MessageDecorator
{
    public function decorate(Message $message): Message
    {
    }
}

手动注册（不带属性和自动配置）的示例

services:
  #...
  App\Decorator\FooDecorator:
    tags:
      -
        name: andreo.eventsauce.message_decorator

消息出盒

安装 andreo/eventsauce-outbox

composer require andreo/eventsauce-outbox

基本配置

andreo_event_sauce:
  #...
  message_outbox:
    enabled: false
    repository:
      doctrine:
        enabled: true
        table_name: message_outbox
    logger: Psr\Log\LoggerInterface # default if monolog bundle has been installed

消费出盒消息

bin/console andreo:eventsauce:message-outbox:consume relay_id

有用的别名

EventSauce\BackOff\BackOffStrategy: EventSauce\BackOff\ExponentialBackOffStrategy

EventSauce\MessageOutbox\RelayCommitStrategy: EventSauce\MessageOutbox\MarkMessagesConsumedOnCommit

快照

要使用

• doctrine snapshot repository
• 版本化的快照
• 条件持久化

需要安装 andreo/eventsauce-snapshotting 包

andreo_event_sauce:
  #...
  snapshot: 
    enabled: false
    repository:
      enabled: false
      doctrine:
        enabled: true
        table_name: snapshot_store
    versioned: false # it enables versioned repository for all aggregates with snapshots enabled
    conditional: false

有用的别名

Andreo\EventSauce\Snapshotting\Repository\Versioned\SnapshotVersionInflector: Andreo\EventSauce\Snapshotting\Repository\Versioned\InflectVersionFromReturnedTypeOfSnapshotStateCreationMethod

Andreo\EventSauce\Snapshotting\Repository\Versioned\SnapshotVersionComparator: Andreo\EventSauce\Snapshotting\Repository\Versioned\EqSnapshotVersionComparator

迁移生成器

安装 andreo/eventsauce-migration-generator

andreo_event_sauce:
  #...
  migration_generator:
    dependency_factory: doctrine.migrations.dependency_factory # default if doctrine migrations bundle has been installed

为 foo 前缀生成迁移

bin/console andreo:eventsauce:doctrine-migrations:generate foo 

有用的别名

EventSauce\MessageRepository\TableSchema\TableSchema: EventSauce\MessageRepository\TableSchema\DefaultTableSchema

聚合

andreo_event_sauce:
  #...
  aggregates:
    foo: # aggregate name
      class: ~ # aggregate FQCN
      repository_alias: fooRepository # according to convention: $name . "Repository"
      message_outbox:
        enabled: false # enable message outbox for this aggregate
        relay_id: foo_aggregate_relay # relay-id for run consume outbox messages command, according to convention: $name . "aggregate_relay"
      dispatchers: [] # dispatcher service aliases (from config, or manually registered), if empty, messages will be sent to all dispatchers
      upcaster: false # enable upcaster for this aggregate
      snapshot:
        conditional: # enable conditional snapshot repository for this aggregate.
          enabled: false
          every_n_event: # you can use this strategy, or make your own implementation
            enabled: false
            number: 100

存储库注入

use Symfony\Component\DependencyInjection\Attribute\Target;
use EventSauce\EventSourcing\AggregateRootRepository;

final class FooHandler {

   public function __construct(
        #[Target('fooRepository')] private AggregateRootRepository $fooRepository
    ){}
}

快照存储库注入（如果启用了聚合快照）

use Symfony\Component\DependencyInjection\Attribute\Target;
use EventSauce\EventSourcing\Snapshotting\AggregateRootRepositoryWithSnapshotting;

final class FooHandler {

   public function __construct(
        #[Target('fooRepository')] private AggregateRootRepositoryWithSnapshotting $fooRepository
    ){}
}

有用的别名

andreo.eventsauce.snapshot.conditional_strategy.$aggregateName: Andreo\EventSauce\Snapshotting\Repository\Conditional\ConditionalSnapshotStrategy

EventSauce\EventSourcing\Serialization\PayloadSerializer: EventSauce\EventSourcing\Serialization\ConstructingPayloadSerializer

EventSauce\EventSourcing\Serialization\MessageSerializer: EventSauce\EventSourcing\Serialization\ConstructingMessageSerializer

EventSauce\UuidEncoding\UuidEncoder: EventSauce\UuidEncoding\BinaryUuidEncoder

EventSauce\EventSourcing\ClassNameInflector: EventSauce\EventSourcing\DotSeparatedSnakeCaseInflector

其他提示

装饰聚合根存储库

<?php
use EventSauce\EventSourcing\AggregateRootId;
use EventSauce\EventSourcing\AggregateRootRepository;
use Symfony\Component\DependencyInjection\Attribute\AsDecorator;

#[AsDecorator(decorates: 'fooRepository')]
final readonly class FooRepository implements AggregateRootRepository
{
    public function __construct(private AggregateRootRepository $regularRepository)
    {
    }

    public function retrieve(AggregateRootId $aggregateRootId): object
    {
        return $this->regularRepository->retrieve($aggregateRootId);
    }

    public function persist(object $aggregateRoot): void
    {
        // ...
    }
    public function persistEvents(AggregateRootId $aggregateRootId, int $aggregateRootVersion, object ...$events): void
    {
        // ...
    }
}

使用此扩展包的示例 API

存储库