<?php

namespace App\Service;

use Prooph\Common\Event\ActionEvent;
use Prooph\EventSourcing\AggregateChanged;
use Prooph\EventStore\ActionEventEmitterEventStore;
use Prooph\EventStore\Plugin\AbstractPlugin;
use Prooph\EventStore\Stream;
use Superbalist\PubSub\PubSubAdapterInterface;

class RedisPlugin extends AbstractPlugin
{
    /**
     * @var PubSubAdapterInterface
     */
    private $pubSub;

    public function __construct(PubSubAdapterInterface $pubSub)
    {
        $this->pubSub = $pubSub;
    }

    public function attachToEventStore(ActionEventEmitterEventStore $eventStore): void
    {
        $this->listenerHandlers[] = $eventStore->attach(ActionEventEmitterEventStore::EVENT_APPEND_TO, function (ActionEvent $event): void {
            /** @var AggregateChanged[] $recordedEvents */
            $recordedEvents = $event->getParam('streamEvents', new \ArrayIterator());

            foreach ($recordedEvents as $recordedEvent) {
                $this->pubSub->publish('event_stored' ,[
                    'event' => \get_class($recordedEvent),
                    'data' => $recordedEvent->toArray()
                ]);
            }
        }, -1);

        $this->listenerHandlers[] = $eventStore->attach(ActionEventEmitterEventStore::EVENT_CREATE, function (ActionEvent $event): void {
            /** @var Stream $stream */
            $stream = $event->getParam('stream');

            foreach ($stream->streamEvents() as $recordedEvent) {
                $this->pubSub->publish('event_stored' ,[
                    'event' => \get_class($recordedEvent),
                    'data' => $recordedEvent->toArray()
                ]);
            }
        }, -1);
    }
}