Netty 4.1 - Writing multiple messages to a single channel

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;

/**
 * Heartbeat server that sends push notification to a peer
 */
public final class HeartBeatServer {

    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx = ServerUtil.buildSslContext();

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final HeartBeatServer serverHandler = new HeartBeatServer();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/**
 * Handler implementation for the HeartBeat server.
 */
@Sharable
public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * @param ctx  channel handler context 
     * @param msg - raw byte buffer
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){
       // Todo - logic to process Object msg
       // Next steps is to obtain the executor service from the 
       // channel handler context

       ctx.channel()
        .eventLoop()
        .scheduleAtFixedRate(() -> {

                ctx.writeAndFlush( Unpooled.copiedBuffer("testing scheduled message\n", Charset.forName("utf-8")))
                        .addListener((ChannelFutureListener) channelFuture -> {
                            // log complete operation evenet
                        });

        }, 5, 5, TimeUnit.SECONDS);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

tx.channel().eventLoop()
        .scheduleAtFixedRate(() -> {

            try {
                FullHttpResponse rsp = new DefaultFullHttpResponse(req.protocolVersion(), OK,
                        Unpooled.wrappedBuffer(CONTENT));
                rsp.headers()
                        .set(CONTENT_TYPE, TEXT_PLAIN)
                        .setInt(CONTENT_LENGTH, rsp.content().readableBytes());
                
              ctx.writeAndFlush(rsp).sync();
            }catch (Exception e) {
                e.printStackTrace();
            }
        }, 3, 3, TimeUnit.SECONDS);

@Override
protected void encode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out) throws Exception {

    final boolean isFull = msg instanceof HttpResponse && msg instanceof LastHttpContent;
    switch (state) {
        case AWAIT_HEADERS: {
            ensureHeaders(msg);

            assert encoder == null;

            final HttpResponse res = (HttpResponse) msg;
            final int code = res.status().code();
            final CharSequence acceptEncoding;

            if (code == CONTINUE_CODE) {
                // We need to not poll the encoding when response with CONTINUE as another response will follow

                // for the issued request. See https://github.com/netty/netty/issues/4079

                acceptEncoding = null;
            } else {
                // Get the list of encodings accepted by the peer.
                acceptEncoding = acceptEncodingQueue.poll();
                if (acceptEncoding == null) {
                    throw new IllegalStateException("cannot send more responses than requests");
                }
            }