[Broker/Client] Close connection if a ping or pong message cannot be sent
* [Broker/Client] Close connection if a ping message cannot be sent
the connection should be closed if a ping message cannot be sent
* Handle write errors for pong messages
@@ -83,7 +83,14 @@ public abstract class PulsarHandler extends PulsarDecoder {
if (log.isDebugEnabled()) {
log.debug("[{}] Replying back to ping message", ctx.channel());
}
- ctx.writeAndFlush(Commands.newPong());
+ ctx.writeAndFlush(Commands.newPong())
+ .addListener(future -> {
+ if (!future.isSuccess()) {
+ log.warn("[{}] Forcing connection to close since cannot send a pong message.",
+ ctx.channel(), future.cause());
+ ctx.close();
+ }
+ });
}
@Override
@@ -110,7 +117,14 @@ public abstract class PulsarHandler extends PulsarDecoder {
log.debug("[{}] Sending ping message", ctx.channel());
}
waitingForPingResponse = true;
- ctx.writeAndFlush(Commands.newPing());
+ ctx.writeAndFlush(Commands.newPing())
+ .addListener(future -> {
+ if (!future.isSuccess()) {
+ log.warn("[{}] Forcing connection to close since cannot send a ping message.",
+ ctx.channel(), future.cause());
+ ctx.close();
+ }
+ });
} else {
if (log.isDebugEnabled()) {
log.debug("[{}] Peer doesn't support keep-alive", ctx.channel());
