src/Rxnet/EventStore/EventStore.php
public function volatileSubscription(string $streamId, bool $resolveLink = false): Observable
    {
        $event = new SubscribeToStream();
        $event->setEventStreamId($streamId);
        $event->setResolveLinkTos($resolveLink);

        return Observable::create(function (ObserverInterface $observer) use ($event) {
            $correlationID = $this->writer->createUUIDIfNeeded();
            $this->writer
                ->composeAndWrite(
                    MessageType::SUBSCRIBE_TO_STREAM,
                    $event,
                    $correlationID
                )
                // When written wait for all responses
                ->merge(
                    $this->readBuffer
                        ->filter(
                            function (SocketMessage $message) use ($correlationID) {
                                // Use same correlationID to pass by this filter
                                return $message->getCorrelationID() == $correlationID;
                            }
                        )
                )
                ->flatMap(
                    function (SocketMessage $message) {
                        $data = $message->getData();

                        if (!$data) {
                            throw new \RuntimeException('Data should not be null in volatile subscription');
                        }

                        switch (get_class($data)) {
                            case SubscriptionDropped::class:
                                return Observable::error(new \Exception("Subscription dropped, for reason : {$data->getReason()}"));
                            case SubscriptionConfirmation::class:
                                return Observable::empty();
                            default:
                                return Observable::of($data);
                        }
                    }
                )
                ->map(
                    function (StreamEventAppeared $eventAppeared) {
                        $record = $eventAppeared->getEvent()->getEvent();
                        /* @var \Rxnet\EventStore\Data\EventRecord $record */

                        return EventRecordFactory::fromEventRecord($record);
                    }
                )
                ->subscribe($observer);

            return new CallbackDisposable(function () {
                $event = new UnsubscribeFromStream();
                $this->writer->composeAndWrite(
                    MessageType::UNSUBSCRIBE_FROM_STREAM,
                    $event
                );
            });
        });
    }
public function volatileSubscription(string $streamId, bool $resolveLink = false): Observable
    {
        $event = new SubscribeToStream();
        $event->setEventStreamId($streamId);
        $event->setResolveLinkTos($resolveLink);

        return Observable::create(function (ObserverInterface $observer) use ($event) {
            $correlationID = $this->writer->createUUIDIfNeeded();
            $this->writer
                ->composeAndWrite(
                    MessageType::SUBSCRIBE_TO_STREAM,
                    $event,
                    $correlationID
                )
                // When written wait for all responses
                ->merge(
                    $this->readBuffer
                        ->filter(
                            function (SocketMessage $message) use ($correlationID) {
                                // Use same correlationID to pass by this filter
                                return $message->getCorrelationID() == $correlationID;
                            }
                        )
                )
                ->flatMap(
                    function (SocketMessage $message) {
                        $data = $message->getData();

                        if (!$data) {
                            throw new \RuntimeException('Data should not be null in volatile subscription');
                        }

                        switch (get_class($data)) {
                            case SubscriptionDropped::class:
                                return Observable::error(new \Exception("Subscription dropped, for reason : {$data->getReason()}"));
                            case SubscriptionConfirmation::class:
                                return Observable::empty();
                            default:
                                return Observable::of($data);
                        }
                    }
                )
                ->map(
                    function (StreamEventAppeared $eventAppeared) {
                        $record = $eventAppeared->getEvent()->getEvent();
                        /* @var \Rxnet\EventStore\Data\EventRecord $record */

                        return EventRecordFactory::fromEventRecord($record);
                    }
                )
                ->subscribe($observer);

            return new CallbackDisposable(function () {
                $event = new UnsubscribeFromStream();
                $this->writer->composeAndWrite(
                    MessageType::UNSUBSCRIBE_FROM_STREAM,
                    $event
                );
            });
        });
    }
public function volatileSubscription ( string $ streamId , bool $ resolveLink = false ) : Observable { $ event = new SubscribeToStream ( ) ; $ event -> setEventStreamId ( $ streamId ) ; $ event -> setResolveLinkTos ( $ resolveLink ) ; return Observable :: create ( function ( ObserverInterface $ observer ) use ( $ event ) { $ correlationID = $ this -> writer -> createUUIDIfNeeded ( ) ; $ this -> writer -> composeAndWrite ( MessageType :: SUBSCRIBE_TO_STREAM , $ event , $ correlationID ) // When written wait for all responses -> merge ( $ this -> readBuffer -> filter ( function ( SocketMessage $ message ) use ( $ correlationID ) { // Use same correlationID to pass by this filter return $ message -> getCorrelationID ( ) == $ correlationID ; } ) ) -> flatMap ( function ( SocketMessage $ message ) { $ data = $ message -> getData ( ) ; if ( ! $ data ) { throw new \ RuntimeException ( 'Data should not be null in volatile subscription' ) ; } switch ( get_class ( $ data ) ) { case SubscriptionDropped :: class : return Observable :: error ( new \ Exception ( "Subscription dropped, for reason : {$data->getReason()}" ) ) ; case SubscriptionConfirmation :: class : return Observable :: empty ( ) ; default : return Observable :: of ( $ data ) ; } } ) -> map ( function ( StreamEventAppeared $ eventAppeared ) { $ record = $ eventAppeared -> getEvent ( ) -> getEvent ( ) ; /* @var \Rxnet\EventStore\Data\EventRecord $record */ return EventRecordFactory :: fromEventRecord ( $ record ) ; } ) -> subscribe ( $ observer ) ; return new CallbackDisposable ( function ( ) { $ event = new UnsubscribeFromStream ( ) ; $ this -> writer -> composeAndWrite ( MessageType :: UNSUBSCRIBE_FROM_STREAM , $ event ) ; } ) ; } ) ; }
This kind of subscription calls a given function for events written after
the subscription is established.

For example, if a stream has 100 events in it when a subscriber connects,
the subscriber can expect to see event number 101 onwards until the time
the subscription is closed or dropped.
This kind of subscription calls a given function for events written after the subscription is established .