src/Rxnet/EventStore/EventStore.php
protected function heartbeat(): DisposableInterface
    {
        return $this->readBuffer
            ->timeout($this->heartBeatRate)
            ->filter(
                function (SocketMessage $message) {
                    return $message->getMessageType()->getType() === MessageType::HEARTBEAT_REQUEST_COMMAND;
                }
            )
            ->subscribe(
                new CallbackObserver(
                    function (SocketMessage $message) {
                        $this->writer->composeAndWrite(MessageType::HEARTBEAT_RESPONSE_COMMAND, null, $message->getCorrelationID());
                    },
                    [$this->connectionSubject, 'onError']
                )
            );
    }
protected function heartbeat(): DisposableInterface
    {
        return $this->readBuffer
            ->timeout($this->heartBeatRate)
            ->filter(
                function (SocketMessage $message) {
                    return $message->getMessageType()->getType() === MessageType::HEARTBEAT_REQUEST_COMMAND;
                }
            )
            ->subscribe(
                new CallbackObserver(
                    function (SocketMessage $message) {
                        $this->writer->composeAndWrite(MessageType::HEARTBEAT_RESPONSE_COMMAND, null, $message->getCorrelationID());
                    },
                    [$this->connectionSubject, 'onError']
                )
            );
    }
protected function heartbeat ( ) : DisposableInterface { return $ this -> readBuffer -> timeout ( $ this -> heartBeatRate ) -> filter ( function ( SocketMessage $ message ) { return $ message -> getMessageType ( ) -> getType ( ) === MessageType :: HEARTBEAT_REQUEST_COMMAND ; } ) -> subscribe ( new CallbackObserver ( function ( SocketMessage $ message ) { $ this -> writer -> composeAndWrite ( MessageType :: HEARTBEAT_RESPONSE_COMMAND , null , $ message -> getCorrelationID ( ) ) ; } , [ $ this -> connectionSubject , 'onError' ] ) ) ; }
Intercept heartbeat message and answer automatically
Intercept heartbeat message and answer automatically