Detaching listener handlers from event store

[webdevilopers]

Since prooph service bus is deprecated we are using the symfony messenger. In order to couple it we copied the publisher from the original bridge but inject the symfony bus:

• https://github.com/prooph/event-store-bus-bridge/blob/master/src/EventPublisher.php

The publisher attaches some listeners for append_to:

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

                if (! $this->inTransaction($eventStore)) {
                    if ($event->getParam('streamNotFound', false)
                        || $event->getParam('concurrencyException', false)
                    ) {
                        return;
                    }

                    foreach ($recordedEvents as $recordedEvent) {
                        $this->eventBus->dispatch($recordedEvent);
                    }
                } else {
                    $this->cachedEventStreams[] = $recordedEvents;
                }
            }
        );

We have a service that is writing directly to the event store and we do not want to publish these events. We can detach the complete plugin from the event store via:

$this->publisher->detachFromEventStore($this->eventStore);

But how could this be achieved using the detach method on the event store itself?

We tried this:

        $publisher = new DefaultListenerHandler(function(ActionEvent $event) {
            dump($event);exit;
        });
        $this->eventStore->detach($publisher);

But we are not sure how to design the closure. In our example nothing happens.

Any help would be appreciated.

Probably related:

• https://github.com/prooph/event-store/issues/232

[prolic]

From the top of my head (on the phone right now): use the return value of the attach call.

On Thu, Aug 6, 2020, 15:30 webDEVILopers notifications@github.com wrote:

Since prooph service bus is deprecated we are using the symfony messenger. In order to use it we copied the bridge from the original bridge but inject the symfony bus:

- https://github.com/prooph/event-store-bus-bridge/blob/master/src/EventPublisher.php

The publisher attaches some listeners for append_to:

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

            if (! $this->inTransaction($eventStore)) {
                if ($event->getParam('streamNotFound', false)
                    || $event->getParam('concurrencyException', false)
                ) {
                    return;
                }

                foreach ($recordedEvents as $recordedEvent) {
                    $this->eventBus->dispatch($recordedEvent);
                }
            } else {
                $this->cachedEventStreams[] = $recordedEvents;
            }
        }
    );

We have a service that is writing directly to the event store and we do not want to publish these events. We can detach the complete plugin from the event store via:

$this->publisher->detachFromEventStore($this->eventStore);

But how could this be achieved using the detach method on the event store itself?

We tried this:

    $publisher = new DefaultListenerHandler(function(ActionEvent $event) {
        dump($event);exit;
    });
    $this->eventStore->detach($publisher);

But we are not sure how to design the closure. In our example nothing happens.

Any help would be appreciated.

— You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub https://github.com/prooph/event-store/issues/410, or unsubscribe https://github.com/notifications/unsubscribe-auth/AADAJPDCGBSDOC4THANTH5LR7MAFRANCNFSM4PW46AEA .

[webdevilopers]

Unfort. the detach method is in a different class than the attach inside the EventPublisher class.

The return value dumped looks like this:

I must confess I don't know how to recreate that closure.

        $publisher = new DefaultListenerHandler(function(ActionEvent $event) {
            // ?!?!?
        });

[webdevilopers]

Any idea - @codeliner maybe?

[unixslayer]

@webdevilopers even if you recreate that closure, you will not be able to detach listener.

$actionEventEmitter = new ProophActionEventEmitter();
$eventStore = new ActionEventEmitterEventStore(new InMemoryEventStore(), $actionEventEmitter);

$callable = function (ActionEvent $event) {};

$eventStore->attach('event_name', $callable);
$eventStore->detach(new DefaultListenerHandler($callable)); // this doesn't detach the listener

This is due to strict comparison in ProophActionEventEmitter

// \Prooph\Common\Event\ProophActionEventEmitter::detachListener
public function detachListener(ListenerHandler $listenerHandler): bool
{
    foreach ($this->events as &$prioritizedListeners) {
        foreach ($prioritizedListeners as &$listenerHandlers) {
            foreach ($listenerHandlers as $index => $listedListenerHandler) {
                if ($listedListenerHandler === $listenerHandler) { // <--- here
                    unset($listenerHandlers[$index]);

                    return true;
                }
            }
        }
    }

    return false;
}

Even with weaker comparison still won't work.

$eventStore->detach(new DefaultListenerHandler(function (ActionEvent $event) {})); // not working
$eventStore->detach(new DefaultListenerHandler($callable)); // might work with loose comparison

The only way to detach the listener is to use value returned from attach call, like @prolic described

$listener = $eventStore->attach('event_name', $callable);
$eventStore->detach($listener); // works like a charm

[webdevilopers]

Thanks @unixslayer for clearing this up. I was really wondering how a closure could be recreated.

In my case the EventPublisher is added via Symfony and the EventStoreBusBridge:

services:
    _defaults:
        public: false

    Prooph\EventStoreBusBridge\EventPublisher:
        class: Rewotec\Common\Infrastructure\Prooph\EventPublisher
        arguments:
            - '@messenger.bus.events'
        tags:
            - { name: 'prooph_event_store.default_store.plugin' }
            - { name: 'prooph_event_store.async_store.plugin' }

So I'm missing the actual part:

$eventStore->attach('event_name', $callable);

That's why I used this solution so far:

$this->eventPublisher->detachFromEventStore($this->eventStore);

This is called in totally different service class, not the configuration of the event store. That's why this is the only reference to the listener I have so far. I guess there is no other way to determine and get the listener then.

[prolic]

I'm closing this.