How to use the eventsourcing.domain.model.events.subscribe function in eventsourcing

def test_publish_subscribe_unsubscribe(self):
        # Check subscribing event handlers with predicates.
        # - when predicate is True, handler should be called
        event = mock.Mock()
        predicate = mock.Mock()
        handler = mock.Mock()

        # Check we can assert there are no event handlers subscribed.
        assert_event_handlers_empty()

        # When predicate is True, handler should be called ONCE.
        subscribe(event_predicate=predicate, subscriber=handler)

        # Check we can assert there are event handlers subscribed.
        self.assertRaises(EventHandlersNotEmptyError, assert_event_handlers_empty)

        # Check what happens when an event is published.
        publish(event)
        predicate.assert_called_once_with(event)
        handler.assert_called_once_with(event)

        # When predicate is True, after unsubscribing, handler should NOT be called again.
        unsubscribe(event_predicate=predicate, subscriber=handler)
        publish(event)
        predicate.assert_called_once_with(event)
        handler.assert_called_once_with(event)

        # Check we can assert there are no event handlers subscribed.

# Define a class that uses the decorator.
        class Aaa(EventSourcedEntity):
            "An event sourced entity."

            def __init__(self, a, *args, **kwargs):
                super(Aaa, self).__init__(*args, **kwargs)
                self._a = a

            @mutableproperty
            def a(self):
                "A mutable event sourced property."

        # Instantiate the class and check assigning to the property publishes an event and updates the object state.
        published_events = []
        subscription = (lambda x: True, lambda x: published_events.append(x))
        subscribe(*subscription)
        entity_id = '1'
        try:
            aaa = Aaa(entity_id=entity_id, entity_version=0, domain_event_id='0', a=1)
            self.assertEqual(aaa.a, 1)
            aaa.a = 'value1'
            self.assertEqual(aaa.a, 'value1')
        finally:
            unsubscribe(*subscription)

        # Check an event was published.
        self.assertEqual(len(published_events), 1)

        # Check the published event was an AttributeChanged event, with the expected attribute values.
        published_event = published_events[0]
        self.assertIsInstance(published_event, Aaa.AttributeChanged)
        self.assertEqual(published_event.name, '_a')

for process_class in self.system.process_classes:
            if not isinstance(process_class, ApplicationWithConcreteInfrastructure):
                if not self.infrastructure_class:
                    raise ProgrammingError("infrastructure_class is not set")
                elif not issubclass(
                    self.infrastructure_class, ApplicationWithConcreteInfrastructure
                ):
                    raise ProgrammingError(
                        "infrastructure_class is not a subclass of {}".format(
                            ApplicationWithConcreteInfrastructure
                        )
                    )

        # Subscribe to broadcast prompts published by a process
        # application in the parent operating system process.
        subscribe(handler=self.broadcast_prompt, predicate=self.is_prompt)

        # Start operating system process.
        for pipeline_id in self.pipeline_ids:
            for process_name, upstream_names in self.system.followings.items():
                process_class = self.system.process_classes[process_name]
                inbox = self.inboxes[(pipeline_id, process_name.lower())]
                outbox = self.outboxes.get((pipeline_id, process_name.lower()))
                os_process = OperatingSystemProcess(
                    application_process_class=process_class,
                    infrastructure_class=self.infrastructure_class,
                    upstream_names=upstream_names,
                    poll_interval=self.poll_interval,
                    pipeline_id=pipeline_id,
                    setup_tables=self.setup_tables,
                    inbox=inbox,
                    outbox=outbox,

def __init__(self):
        self.projection: TicketListProjection = TicketListProjection()
        subscribe(self.add_ticket, self.is_ticket_created)
        subscribe(self.update_ticket, self.is_ticket_updated)
        subscribe(self.delete_ticket, self.is_ticket_deleted)

def __init__(self):
        self.projection: TicketListProjection = TicketListProjection()
        subscribe(self.add_ticket, self.is_ticket_created)
        subscribe(self.update_ticket, self.is_ticket_updated)
        subscribe(self.delete_ticket, self.is_ticket_deleted)

process_class = msg.process_application_class
        if msg.infrastructure_class:
            process_class = process_class.mixin(msg.infrastructure_class)

        # Reset the database connection (for Django).
        process_class.reset_connection_after_forking()

        # Construct the process application.
        self.process = process_class(pipeline_id=self.pipeline_id)
        assert isinstance(self.process, ProcessApplication)

        # Subscribe the slave actor's send_prompt() method.
        #  - the process application will call publish_prompt()
        #    and the actor will receive the prompt and send it
        #    as a message.
        subscribe(predicate=self.is_my_prompt, handler=self.send_prompt)

        # Close the process application persistence policy.
        #  - slave actor process application doesn't publish
        #    events, so we don't need this
        self.process.persistence_policy.close()

        # Unsubscribe process application's publish_prompt().
        #  - slave actor process application doesn't publish
        #    events, so we don't need this
        unsubscribe(
            predicate=self.process.persistence_policy.is_event,
            handler=self.process.publish_prompt,
        )

        # Construct and follow upstream notification logs.
        for upstream_application_name in self.upstream_application_names:

def wrap(func):
        def handler(event):
            if isinstance(event, (list, tuple)):
                # Call handler once for each event.
                for e in event:
                    handler(e)
            elif not args or isinstance(event, args):
                # Call handler if there are no classes or have an instance.
                func(event)

        subscribe(handler=handler, predicate=lambda _: True)
        return func

self.notification_state_key = (
                self._event_store.record_manager.field_names.state
            )

        # Publish prompts for any domain events that we persist.
        #  - this means when a process application is used as a simple
        #    application, with calls to aggregate __save__() methods,
        #    then the new events that are stored might need to be processed
        #    by another application. So it can help also to publish a
        #    prompt after the events have been stored so that followers
        #    can be immediately nudged to process the events (otherwise they
        #    would only pick up the events next time they happen to run).
        #    The particular way a prompt published here is actually sent to
        #    any followers is the responsibility of a particular system runner.
        if self._persistence_policy:
            subscribe(
                predicate=self._persistence_policy.is_event, handler=self.publish_prompt
            )

assert len(self.processes) == 0, "Already running"

        # Construct the processes.
        for process_class in self.system.process_classes.values():
            self._construct_app_by_class(process_class)

        # Tell each process about the processes it follows.
        for followed_name, followers in self.system.followers.items():
            followed = self.processes[followed_name]
            followed_log = followed.notification_log
            for follower_name in followers:
                follower = self.processes[follower_name]
                follower.follow(followed_name, followed_log)

        # Do something to propagate prompts.
        subscribe(predicate=is_prompt, handler=self.handle_prompts)