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 ) ; } } }
Helper method to create write and flush the keepalive message .