nsq/nsqphp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

handle heartbeat after reading frame

Browse Source
This commit is contained in:
Konstantin Grachev
2021-02-05 00:08:06 +03:00
parent 57b7715c2e
commit 443473f53d
2 changed files with 12 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/Connection.php
View File

@ -189,7 +189,7 @@ abstract class Connection
$buffer = $this->buffer->append($bytes);
return match ($type = $buffer->consumeUint32()) {
$frame = match ($type = $buffer->consumeUint32()) {
0 => new Response($buffer->flush()),
1 => new Error($buffer->flush()),
2 => new Message(
@ -201,6 +201,14 @@ abstract class Connection
),
default => throw new NsqException('Unexpected frame type: '.$type)
};
if ($frame instanceof Response && $frame->isHeartBeat()) {
yield $this->command('NOP');
return $this->readFrame();
}
return $frame;
});
}
@ -213,17 +221,11 @@ abstract class Connection
/** @var Response $response */
$response = yield $this->readResponse();
if ($response->isHeartBeat()) {
yield $this->command('NOP');
return $this->checkIsOK();
}
if (!$response->isOk()) {
throw new BadResponse($response);
}
$this->logger->info('Ok checked.');
$this->logger->debug('Ok checked.');
return call(static function (): void {});
});

10
src/Consumer.php
View File

@ -63,21 +63,15 @@ final class Consumer extends Connection
return $frame;
}
if ($frame instanceof Response && $frame->isHeartBeat()) {
yield $this->command('NOP');
return $this->readMessage();
}
if ($frame instanceof Error) {
if ($frame->type->terminateConnection) {
yield $this->close();
}
return new Failure(new NsqError($frame));
throw new NsqError($frame);
}
return new Failure(new NsqException('Unreachable statement.'));
throw new NsqException('Unreachable statement.');
});
}