#NettyFramework - Echo Server, Client 구현

Echo Server

package netty_framework;
 
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
 
import javax.swing.plaf.synth.SynthProgressBarUI;
 
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;
 
public class EchoServer {
 
    // 서버 소켓 포트 번호를 지정합니다.
    //private static final int PORT = 30081;
 
    public static void main(String[] args) {
    /*
        NioEventLoop는 I/O 동작을 다루는 멀티스레드 이벤트 루프입니다.
        네티는 다양한 이벤트 루프를 제공합니다.
        이 예제에서는 두개의 Nio 이벤트 루프를 사용합니다.
        첫번째 'parent' 그룹은 인커밍 커넥션(incomming connection)을 액세스합니다.
        두번째 'child' 그룹은 액세스한 커넥션의 트래픽을 처리합니다.
        만들어진 채널에 매핑하고 스레드를 얼마나 사용할지는 EventLoopGroup 구현에 의존합니다.
        그리고 생성자를 통해서도 구성할 수 있습니다.
    */
        EventLoopGroup parentGroup = new NioEventLoopGroup(1);
        EventLoopGroup childGroup = new NioEventLoopGroup();
        try{
            // 서버 부트스트랩을 만듭니다. 이 클래스는 일종의 헬퍼 클래스입니다.
            // 이 클래스를 사용하면 서버에서 Channel을 직접 세팅 할 수 있습니다.
            
            ServerBootstrap sb = new ServerBootstrap();
        
            sb.group(parentGroup, childGroup)
            // 인커밍 커넥션을 액세스하기 위해 새로운 채널을 객체화 하는 클래스 지정합니다.
            .channel(NioServerSocketChannel.class)
            // 상세한 Channel 구현을 위해 옵션을 지정할 수 있습니다.
            //동시에 수용 가능한 클라이언트 연결 요청 개수를 서버 소켓에 설정 가능한 옵션
            .option(ChannelOption.SO_BACKLOG, 100)
            .handler(new LoggingHandler(LogLevel.INFO))
            // 새롭게 액세스된 Channel을 처리합니다.
            
            // ChannelInitializer는 특별한 핸들러로 새로운 Channel의
            // 환경 구성을 도와 주는 것이 목적입니다.
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    ChannelPipeline cp = sc.pipeline();
                    cp.addLast(new EchoServerHandler());
 
                }
            });
            
            //SocketAddress address = new InetSocketAddress("169.254.40.218",10081);
    //        SocketAddress address2 = new InetSocketAddress("172.10.11.20",10081);
            
            // 인커밍 커넥션을 액세스하기 위해 바인드하고 시작합니다.
            ChannelFuture cf = sb.bind(8080).sync();
//            cf = sb.bind(10087).sync();
            
            // 서버 소켓이 닫힐때까지 대기합니다.
            cf.channel().closeFuture().sync();
        //    cf2.channel().closeFuture().sync();
        }catch(Exception e){
            e.printStackTrace();
        }
        finally{
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

Server Inbound Handler

 
package netty_framework;
 
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
 
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
 
    // 채널을 읽을 때 동작할 코드를 정의 합니다.
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server read : " + msg);
        System.out.println(ctx.channel().localAddress().toString() );
        ctx.write(msg); // 메시지를 그대로 다시 write 합니다.
    }
 
    // 채널 읽는 것을 완료했을 때 동작할 코드를 정의 합니다.
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush(); // 컨텍스트의 내용을 플러쉬합니다.
    };
 
    // 예외가 발생할 때 동작할 코드를 정의 합니다.
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace(); // 쌓여있는 트레이스를 출력합니다.
        ctx.close(); // 컨텍스트를 종료시킵니다.
    }
}

Client

package netty_framework;
 
import io.netty.bootstrap.Bootstrap;
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.NioSocketChannel;
 
public class EchoClient {
    // 호스트를 정의합니다. 로컬 루프백 주소를 지정합니다.
    private static final String HOST = "localhost";
    // 접속할 포트를 정의합니다.
    private static final int PORT = 8080;
    // 메시지 사이즈를 결정합니다.
    static final int MESSAGE_SIZE = 256;
 
    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
 
        try{
            Bootstrap b = new Bootstrap();
            b.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    ChannelPipeline cp = sc.pipeline();
                    cp.addLast(new EchoClientHandler());
                }
            });
 
            ChannelFuture cf = b.connect(HOST, PORT).sync();
            cf.channel().closeFuture().sync();
        }
        catch(Exception e){
            e.printStackTrace();
        }
        finally{
            group.shutdownGracefully();
        }
    }
}

Client Inbound Handler

package netty_framework;
 
import java.util.Arrays;
 
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
 
public class EchoClientHandler extends ChannelInboundHandlerAdapter{
    private final ByteBuf message;
 
    // 초기화
    public EchoClientHandler(){
        message = Unpooled.buffer(EchoClient.MESSAGE_SIZE);
        // 예제로 사용할 바이트 배열을 만듭니다.
        byte[] str = "abcefg".getBytes();
        // 예제 바이트 배열을 메시지에 씁니다.
        message.writeBytes(str);
        message.writeByte((byte)0x11);
 
    }
 
    // 채널이 활성화 되면 동작할 코드를 정의합니다.
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 메시지를 쓴 후 플러쉬합니다.
        ctx.writeAndFlush(message);
    }
    
    public byte[] getMessageID(byte[] body) {
 
        int readCount = 0;
        // ByteBuf buf = Unpooled.buffer(body.length);
        int bodySize = body.length;
 
        for (int i = 0; i < bodySize; i++) {
            if (body[i] == (byte)0x11) {
                break;
            }
            readCount++;
        }
 
        byte[] messageID = new byte[readCount];
 
        messageID = Arrays.copyOfRange(body, 0, readCount);
 
        return messageID;
 
    }
    
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
        // 받은 메시지를 ByteBuf형으로 캐스팅합니다.
        ByteBuf byteBufMessage = (ByteBuf) msg;
        // 읽을 수 있는 바이트의 길이를 가져옵니다.
        int size = byteBufMessage.readableBytes();
 
        // 읽을 수 있는 바이트의 길이만큼 바이트 배열을 초기화합니다.
        byte [] byteMessage = new byte[size];
        // for문을 돌며 가져온 바이트 값을 연결합니다.
        for(int i = 0 ; i < size; i++){
            byteMessage[i] = byteBufMessage.getByte(i);
        }
        byte[] message =getMessageID(byteMessage);
        System.err.println(new String(message));
        // 바이트를 String 형으로 변환합니다.
//        String str = new String(byteMessage);
 
        // 결과를 콘솔에 출력합니다.
//        System.out.println(str);
 
        // 그후 컨텍스트를 종료합니다.
        ctx.close();
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}