netty time server/client sample in scala

netty-timeclient.scala

package netty.sample

import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.buffer.ByteBuf
import io.netty.bootstrap.Bootstrap
import io.netty.channel._
import io.netty.channel.nio._
import io.netty.channel.socket._
import io.netty.channel.socket.nio._
import io.netty.handler.codec._

class TimeClientHandler extends ChannelInboundHandlerAdapter {

  override def channelRead(ctx: ChannelHandlerContext, msg: Object) {
    val time = msg.asInstanceOf[UnixTime]
    println("server time " + time)
    ctx.close
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
    cause.printStackTrace
    ctx.close
  }
}

class TimeDecoder extends ByteToMessageDecoder {
  override def decode(ctx: ChannelHandlerContext, in: ByteBuf, out: java.util.List[Object]) = {
    if (in.readableBytes >= 4) {
      out.add(new UnixTime(in.readInt))
    }
  }
}

object TimeClient {

  def main(args: Array[String]) {
  val workerGroup = new NioEventLoopGroup
    try {
      val boot = new Bootstrap()
      boot.group(workerGroup)
        .channel((classOf[NioSocketChannel]))
        .handler(new ChannelInitializer[SocketChannel]() {
          def initChannel(ch: SocketChannel) {
            ch.pipeline().addLast(new TimeDecoder, new TimeClientHandler())
          } 
        })
        .option[java.lang.Boolean](ChannelOption.SO_KEEPALIVE, true)

      println("connect to server")
      val f: ChannelFuture  = boot.connect("localhost", 8000).sync()
      f.channel().closeFuture().sync()
      println("client exit")
    } finally {
	  workerGroup.shutdownGracefully()
    }
  }
}

netty-timeserver-sample.scala

package netty.sample

import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.buffer.ByteBuf
import io.netty.bootstrap.ServerBootstrap
import io.netty.channel._
import io.netty.channel.nio._
import io.netty.channel.socket._
import io.netty.channel.socket.nio._
import io.netty.handler.codec._

class TimeServerHandler extends ChannelInboundHandlerAdapter {

  override def channelActive(ctx: ChannelHandlerContext) {
    println("accept a new client")
    ctx.writeAndFlush(new UnixTime)
      .addListener(ChannelFutureListener.CLOSE)
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: Object) {
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
    cause.printStackTrace
    ctx.close
  }
}

object TimeServer {

  def main(args: Array[String]) {
  val bossGroup = new NioEventLoopGroup
	val workerGroup = new NioEventLoopGroup
    try {
      val boot = new ServerBootstrap()
      boot.group(bossGroup, workerGroup)
        .channel((classOf[NioServerSocketChannel]))
        .childHandler(new ChannelInitializer[SocketChannel]() {
          def initChannel(ch: SocketChannel) {
            // add the lower level first
            ch.pipeline().addLast(new TimeEncoder(), new TimeServerHandler())
          } 
        })
        .option[java.lang.Integer](ChannelOption.SO_BACKLOG, 128)        
        .childOption[java.lang.Boolean](ChannelOption.SO_KEEPALIVE, true)

      // Bind and start to accept incoming connections.
      val f: ChannelFuture  = boot.bind(8000).sync(); 
      // Wait until the server socket is closed.
      // In this example, this does not happen, but you can do that to gracefully
      // shut down your server.
      f.channel().closeFuture().sync();
    } finally {
	  workerGroup.shutdownGracefully()
	  bossGroup.shutdownGracefully()
    }
  }
}

class TimeEncoder extends MessageToByteEncoder[UnixTime] {
  def encode(ctx: ChannelHandlerContext, msg: UnixTime, out: ByteBuf) {
    println("encode time value " + msg.t) 
    out.writeInt(msg.t)
  }
}