code: private void createAndWriteKeepAlive(final ChannelHandlerContext ctx) {
          final CouchbaseRequest keepAlive = createKeepAliveRequest();
          if (keepAlive != null) {
              Subscriber<CouchbaseResponse> subscriber = new KeepAliveResponseAction(ctx);
              keepAlive.subscriber(subscriber);
              keepAlive
                  .observable()
                  .timeout(env().keepAliveTimeout(), TimeUnit.MILLISECONDS)
                  .subscribe(subscriber);
  
              onKeepAliveFired(ctx, keepAlive);
  
              Channel channel = ctx.channel();
              if (channel.isActive() && channel.isWritable()) {
                  ctx.pipeline().writeAndFlush(keepAlive);
              }
          }
      }
docs: Helper method to create, write and flush the keepalive message.