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']
                )
            );
    }