Netty(5) - TCP/IP 파일 송수신

web 을 주로 하다보면 network 프로그래밍을 할 경우가 많지 않다.

spring 을 사용하다 보니 필요 시 spring integration 을 이용하거나

아주 간단할 경우 Socket을 직접 만들어 레거시한 코드를 작성하곤 한다.

 

Netty는 네트워크 프로그래밍을 쉽고 간편하게 그리고 유연하게 할 수 있도록 도와주는 Framework 이다.

강력한 비동기 이벤트 드리븐 기반으로 동작하기 때문에 개발자는 Business Logic 에만 집중할 수 있다.

​

Netty 에 대한 기초공부와 Handler, Encoder, Decoder 등을 간단하게 살펴보고

File 을 주고받는 application을 개발해봤다.

​

간단히 파일만 주고 받는 것이 아니라

전문통신을 통해 파일과 더불어 사용자 정보 등 필요한 정보를 짜여진 protocol 에 맞춰 통신해야 해서

ByteArrayDecoder 를 선택하고 handler 에서 ByteBuf msg 를 byte[] 로 받아 dto 에서 parsing 하여

file 을 저장하는 방식으로 개발을 진행했다.

​

아래 코드와 같이 handler 에서 read 시 byte array로 변경하고 dto 에 넘겨서 parsing 하고 저장

parsing이 끝난 객체 정보를 scheduler에 add 해서 별도 분석 진행.

@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
  log.info("receive file"); 

  byte[] bytes = (byte[])msg; 
  PacketDto packet = PacketDto.builder().bytes(bytes).filePath(prop.getFilePath()).build(); 
  analyzeSched.addQueue(packet); 
  
  log.info("save file complete"); 
}

잘 돌아가는 것 같지만

송신하는 파일 사이즈가 조금만 커지면 문제가 발생하는 코드이다.

​

왜그럴까?

​

TCP 프로토콜은 Stream-Oriented 하다.

발신자가 "ABC"를 보내고 "DEF"를 보내면 TCP는 두개의 독립된 메시지로 인식하지 않고

"ABCDEF"라는 하나의 Stream으로 인식한다.

TCP는 상황에 따라 한번에 "ABCDEF"를 보낼 수도 있고, "ABCD", "EF"로 보낼 수도 있다.

 

즉, 위의 ChannelRead method 에는 상황에 따라 여러번에 걸쳐 데이터가 read 될 수 있다는 것이다.

(실제로 사이즈를 조금 크게 한 파일을 전송해 보면 channelRead 의 "receive file" 이라는 log가 여러번 찍힘)

​

netty의 user-guide 문서에도 이 부분에 대한 설명이 있다.

https://netty.io/wiki/user-guide-for-4.x.html

​

netty 에서는 이러한 처리를 자동으로 해 줄 수 있는 다양한 decoder들을 제공한다.

StringDecoder를 보면 (https://netty.io/4.0/api/io/netty/handler/codec/string/StringDecoder.html)

Please note that this decoder must be used with a proper ByteToMessageDecoder such as DelimiterBasedFrameDecoder or LineBasedFrameDecoder if you are using a stream-based transport such as TCP/IP. A typical setup for a text-based line protocol in a TCP/IP socket would be:

​라는 메시지를 볼 수 있다.

@ChannelHandler.Sharable 
public class StringDecoder extends MessageToMessageDecoder<ByteBuf> 
Decodes a received ByteBuf into a String. 
ChannelPipeline pipeline = ...; 

// Decoders 
pipeline.addLast("frameDecoder", new LineBasedFrameDecoder(80)); 
pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8)); 

// Encoder 
pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));

StringDecoder 앞에 그냥 ByteToMessageDecoder를 add 하는것이 아니라

채팅과 같은 application을 만들 때 개행문자로 끝난다는 약속이 있다면

LineBasedFrameDecoder를 pipeline 가장 앞에 추가해 줘야 한다는 것이다.

​

이제 본문으로 돌아와서 파일을 수신할 때 이러한 stream 처리를 하기 위해서 여러 방법이 있을 수 있다.

나는 handler를 이렇게 수정하여 처리 했다.

​

1. 모든 stream을 받기 위한 buffer 변수 추가

private ByteBuf buffer; 
private int totalSize = 0;

2. buffer 초기화

@Override 
public void handlerAdded(ChannelHandlerContext ctx) throws Exception { 
  log.info("handler added"); 
  buffer = ctx.alloc().buffer(); 
}

3. buffer clear

@Override 
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { 
  log.info("handler removed"); 
  clearBuffer(); 
} 

private void clearBuffer() { 
  if (buffer != null) { 
    buffer.release(); 
    buffer = null; 
  } 
}

4. Channel read

@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
  ByteBuf b = (ByteBuf)msg; 
  buffer.writeBytes(b); 
  b.release(); 
  
  int readableByte = buffer.readableBytes(); 
  if (readableByte > 12 && totalSize == 0) { 
    byte[] bytes = new byte[4]; 
    buffer.getBytes(8, bytes); 
    this.totalSize = CommonUtils.bytesToInt(bytes); 
  } 
  
  if (readableByte == totalSize) { 
    byte[] bytes = new byte[buffer.readableBytes()]; 
    buffer.readBytes(bytes); 
    PacketDto packet = PacketDto.builder()
                                .bytes(bytes)
                                .filePath(commonProp.getFilePath())
                                .build(); 
    analyzeSched.addQueue(packet); 
    clearBuffer(); 
  } 
}

buffer 에 byte를 write 하고 읽을 수 있는 byte size를 확인하고

현재 프로토콜에서 header 정보로 - magic code 4 byte + version 4 byte + total size 4 byte - 를 사용하기 때문에

위 코드처럼 total size를 얻어온다.

readableByte == toalSize (모두 읽었다면) business logic 처리 하고 buffer 를 clear 한다.

​

잘 동작 한다. 굳~