TCP 소켓 통신 중 클라이언트의 데이터 혼선에 관한 문의 입니다.

[Jeonghee Song]

안녕하세요. 이희승님 덕분에 소켓 통신을 기초 지식 매우 부족하지만 쉽게 구현하고 있는 사람입니다.

다름이 아니고, 소켓 통신중 몇일 동안 고민하고 시도해 보아도 잘 풀리지 않는 점이 있어 이렇게 문의를 드려봅니다.

현재 제가 구현하고 있는 방식은 Camel 서버에 netty4 컴포넌트를 올려 한 IP에 임의의 포트를 여는 서버를 만들어 채팅 서버입니다.

데이터 흐름은 클라이언트가 -> 서버 로 약속된 데이터

(데이터 구조는 다음과 같습니다=> 시작문자+ BASE64로 인코딩된 컨텐츠 부+ 끝문자)를

전송하면 서버에서 ByteArrayDecoder를 상속받은 CustomDecoder에서 데이터의 처음이 시작문자, 데이터의 끝이 끝문자 면 다음 핸들러로 넘겨서

비즈니스 로직을 처리하고 다시 리턴해주는 형태입니다. 사실 Decoder를 무엇을 써야 하는지 잘 몰라서 ByteArrayDecoder를 사용했습니다...

그런데 이것이 실제로 데이터가 실제로는 중간에 끊어져서 오기도하고 다음 데이터랑 합쳐져서 오기도 해서 끝 문자가 안왔을땐 임시로 저장했다가

끝문자가 오면 끝문자 부분까지만 잘라서 넘기고 나머지는 다시 임시저장하는 형태로 작성하느라 몇일 애먹었습니다. ^^;;

이것까지는 문제가 없었으나 , 파일전송을 처리하면서 문제가 발생했습니다.

파일전송은 클라이언트에서 파일을 16000바이트로 잘라서 컨텐츠 부에 담아 서버로 전송하면 서버가 해당 클라이언트에게 전달하고,

해당 클라이언트는 다음 파일을 보내라~ 하는 방식으로 파일을 전송 받는 구조인데요.

이것이 문제가 A가 B클라이언트에게 파일 데이터를 전송하는 중에 B클라이언트가 A클라이언트에 메세지 전송을 초당 2회 정도의 속도로 빠르게

보내면 서버 CustomDecoder에서 A가 B에게 보내는 데이터와 B가 A에게 보내는 데이터에 혼선이 생겨 데이터를 완성된 형태로 편집 할 수가 없습니다.

이러한 형태 입니다.

        ABCDEF 전송중
A  ---------------------------------->   |
                                           |
       123456 전송중               |  서버         ABCD / 123 / EF / 56  전송받음    
B ----------------------------------->   |


이러한 상태를 어찌 해결할수 있을까요~ 고견부탁드립니다 ~~

[이희승 (Trustin Lee)]

A 와 B 는 서로 다른 커넥션이므로 커넥션별로 다른 디코더 인스턴스를 사용하면 될 것 같아 보입니다.  ChannelInitializer 에서 매번 새로운 디코더 인스턴스를 생성해서 파이프라인어 넣어 주면 어떨까요?

 

--

https://twitter.com/trustin

https://twitter.com/trustin_ko

https://twitter.com/netty_project

--

이 메일은 Google 그룹스 'Netty Korean User Group' 그룹에 가입한 분들에게 전송되는 메시지입니다.

이 그룹에서 탈퇴하고 더 이상 이메일을 받지 않으려면 netty-ko+u...@googlegroups.com에 이메일을 보내세요.

웹에서 이 토론을 보려면 https://groups.google.com/d/msgid/netty-ko/a8d4c08f-877c-4725-a687-aa526ac136cb%40googlegroups.com을(를) 방문하세요.

더 많은 옵션을 보려면 https://groups.google.com/d/optout을(를) 방문하세요.

 

[김남철]

디코더 구현 부분이 문제가 있을것으로 판단되네요.

해당 디코더의 소스를 올리시면 도움을 받으실것으로 보입니다.

일반적으로 패킷 구성은

시작문자 + 패킷의 전체 사이즈 + 바디부로 구성합니다.

네티 디코더에서는 이 구조에 맞게 패킷이 다 들어오기 전까지 return; 해버리면 ByteBuf에 담기게 됩니다.

채널간의 데이타 혼선도 일단 디코더 구현부부터 수정하고 다시 테스트 해보는게 나을듯 합니다.

2015년 1월 22일 오후 2:53, Jeonghee Song <mbhv...@gmail.com>님이 작성:

--

[Jeonghee Song]

소스코드는 이러합니다~


import java.util.Arrays;
import java.util.List;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.bytes.ByteArrayDecoder;

@Sharable
public class CustomDataDecoder extends ByteArrayDecoder{
   
    // 단편화 데이터 임시 저장 버퍼
    private static ByteBuf incompleteBuf = Unpooled.buffer();

    // 단편화 여부 확인값
    private static boolean isIncomplete = false;
   
    Logger logger = LogManager.getLogger(this.getClass());
   
    public CustomDataDecoder() {
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
       
        ByteBuf buffer = Unpooled.buffer();
       
        buffer = (ByteBuf)msg;
       
        int length = buffer.readableBytes();
        byte[] bytes = new byte[length];
       
        while(buffer.isReadable()){
            bytes[buffer.readerIndex()] = buffer.readByte();
        }
   
        // 데이터 시작 추가된 구분자 |
        char startCode = 124;
        byte delimterByte = (byte)startCode;
               
        // 데이터 마지막 부분 구분자
        char lastCode = 127;
        byte lastCharCode = (byte)lastCode;
                       
        boolean hasStartDelimiter = false;
        boolean hasEndDelimiter = false;
       
       
        // 패킷 분석 시작
       
        // 데이터 시작 끝 포함 여부 확인
        for (int i = 0 ; i < bytes.length; i++){
           
            if(bytes[i] == delimterByte){
                hasStartDelimiter = true;   
            } else if(bytes[i] == lastCharCode){
                hasEndDelimiter = true;
            }
        }
       
        // 완전한 상태의 패킷 일 경우 바로 그냥 전송
        if(bytes[0] == delimterByte && bytes[bytes.length-1] == lastCharCode){
   
            ByteBuf returnByte = Unpooled.copiedBuffer(bytes);
            out.add(returnByte);
           
            //System.out.println("CASE 0");
           
        // 완전하지 않은 상태의 패킷 CASE 1 : TEMP 데이터가 없을 경우 (첫 데이터)   
        } else if(bytes[0] == delimterByte && bytes[bytes.length-1] != lastCharCode && !isIncomplete){
           
            incompleteBuf.writeBytes(bytes);
            isIncomplete = true;
           
            //System.out.println("CASE 1");
       
        // 완전하지 않은 상태의 패킷 CASE 2 : TEMP 데이터가 있는 경우
        // TEMP 데이터가 있는데 새로운 시작점이 들어온 경우   
        } else if(bytes[0] == delimterByte && bytes[bytes.length-1] != lastCharCode && isIncomplete){
           
            if(hasEndDelimiter){
               
                int endIndex = 0;
               
                // 데이터 끝문자 포함 여부 및 위치 찾기
                for (int i = 0 ; i < bytes.length; i++){
                   
                    if(bytes[i] == lastCharCode){
                        endIndex = i;
                    }
                }
               
                byte[] releaseData = Arrays.copyOfRange(bytes, 0, endIndex+1);
                ByteBuf returnByte = Unpooled.copiedBuffer(releaseData);
                out.add(returnByte);
               
                byte[] remainData = Arrays.copyOfRange(bytes, endIndex+1, bytes.length);
               
                incompleteBuf.writeBytes(remainData);
               
            } else {
               
                incompleteBuf.writeBytes(bytes);
            }
           
            //System.out.println("CASE 2");
           
       
        // 완전하지 않은 상태의 패킷 CASE 3 : TEMP 데이터가 있을 경우 (끝부분은 delimiter 존재)
        } else if(bytes[0] != delimterByte && bytes[bytes.length-1] == lastCharCode && isIncomplete){
           
            incompleteBuf.writeBytes(bytes);
           
            byte[] tempBytes = new byte[incompleteBuf.readableBytes()];
            int readerIndex = incompleteBuf.readerIndex();
            incompleteBuf.getBytes(readerIndex, tempBytes);
           
            int lastStartIndex = 0;
            int startCount = 0;
            int endCount = 0;
           
            // TempByte의 데이터 체크
            for (int i = 0 ; i < tempBytes.length; i++){
               
                // 마지막 시작점 취득
                if(tempBytes[i] == delimterByte){
       
                    lastStartIndex = i;
                    startCount++;
                   
                } else if(tempBytes[i] == lastCharCode){
                    endCount++;
                }
               
            }
       
            // 시작점 종료점 카운트가 같으면 전부 전송
            if(startCount == endCount){
               
                ByteBuf returnByte = Unpooled.copiedBuffer(tempBytes);
                out.add(returnByte);
               
                incompleteBuf.discardReadBytes();
                incompleteBuf.resetReaderIndex();
                incompleteBuf.resetWriterIndex();
                isIncomplete = false;
           
            // 시작점 종료점 카운트가 같지 않으면 뒤에서 부터 추출하여 완성된 부분 전송
            } else {
               
                // 마지막 완성된 형태 커맨드 추출
                byte[] releaseData = Arrays.copyOfRange(tempBytes, lastStartIndex, tempBytes.length);
               
               
                if(tempBytes[lastStartIndex-1] == lastCharCode){
                   
                    byte[] lastRemainData = Arrays.copyOfRange(tempBytes, 0, lastStartIndex);
                   
                    int middleStartIndex = 0;
                   
                    // TempByte의 데이터 체크
                    for (int i = 0 ; i < lastRemainData.length; i++){
                       
                        // 마지막 시작점 취득
                        if(lastRemainData[i] == delimterByte){
                           
                            middleStartIndex = i;

                        }
                    }
                   
                    byte[] secondReleaseData = Arrays.copyOfRange(lastRemainData, middleStartIndex, lastRemainData.length);
                   
                    byte[] concatBytes = ArrayUtils.addAll(releaseData,secondReleaseData);
                   
                    incompleteBuf.discardReadBytes();
                    incompleteBuf.resetReaderIndex();
                    incompleteBuf.resetWriterIndex();
                   
                    ByteBuf returnByte = Unpooled.copiedBuffer(concatBytes);
                    out.add(returnByte);
                   
                    byte[] remainData = Arrays.copyOfRange(tempBytes, 0, middleStartIndex);
                   
                    incompleteBuf.writeBytes(remainData);
                   
                   
                } else {
                   
                    incompleteBuf.discardReadBytes();
                    incompleteBuf.resetReaderIndex();
                    incompleteBuf.resetWriterIndex();
                   
                    ByteBuf returnByte = Unpooled.copiedBuffer(releaseData);
                    out.add(returnByte);
                   
                    byte[] remainData = Arrays.copyOfRange(tempBytes, 0, lastStartIndex);
                   
                    incompleteBuf.writeBytes(remainData);
                }   
               
            }
           
            //System.out.println("CASE 3");
           
           
        // 완전하지 않은 상태의 패킷 CASE 4 : 중간에 delimiter가 존재
        } else if(hasEndDelimiter && bytes[bytes.length-1] != lastCharCode && isIncomplete){
                       
            incompleteBuf.writeBytes(bytes);
           
            byte[] tempBytes = new byte[incompleteBuf.readableBytes()];
            int readerIndex = incompleteBuf.readerIndex();
            incompleteBuf.getBytes(readerIndex, tempBytes);
           
            incompleteBuf.discardReadBytes();
            incompleteBuf.resetReaderIndex();
            incompleteBuf.resetWriterIndex();
           
            int startDelimiterIndex = 0;
            int endDelimiterIndex = 0;
           
            boolean alreadyCheck = false;
           
            // 서버 접속 코드 포함 여부 및 위치 찾기
            for (int i = 0 ; i < tempBytes.length; i++){
               
                if(tempBytes[i] == delimterByte && !alreadyCheck){
                    startDelimiterIndex = i;
                    alreadyCheck = true;
                   
                } else if(tempBytes[i] == lastCharCode){
                    endDelimiterIndex = i;
                }
            }
           

            byte[] releaseData = Arrays.copyOfRange(tempBytes, startDelimiterIndex, endDelimiterIndex+1);
            ByteBuf returnByte = Unpooled.copiedBuffer(releaseData);
            out.add(returnByte);
           
            // 처음이 데이터 시작 부분이면 마지막 데이터 끝 구분자 이후만 버퍼에 저장
            if(startDelimiterIndex == 0){
               
                byte[] remainData = Arrays.copyOfRange(tempBytes, endDelimiterIndex+1, tempBytes.length);
               
                incompleteBuf.writeBytes(remainData);
               
            } else {
               
                byte[] firstRemainData = Arrays.copyOfRange(tempBytes, 0, startDelimiterIndex);
                byte[] lastRemainData = Arrays.copyOfRange(tempBytes, endDelimiterIndex+1, tempBytes.length);
               
                byte[] concatBytes = ArrayUtils.addAll(firstRemainData,lastRemainData);
               
                incompleteBuf.writeBytes(concatBytes);
            }
           
            //System.out.println("CASE 4");
   
        // 완전하지 않은 상태의 패킷 CASE 5 : 첫부분과 끝부분이 없음   
        } else if(!hasStartDelimiter && !hasEndDelimiter && isIncomplete){
           
            incompleteBuf.writeBytes(bytes);
           
            //System.out.println("CASE 5");
       
        } else {
       
            String tempBuffer = new String(bytes, "UTF-8").trim();
           
            logger.debug("Unknown Message");
            logger.debug(tempBuffer);
            //System.out.println("Unknown Case");
        }
       
       
    }
}   

[Jeonghee Song]

답변 감사합니다.

커넥션 별로 디코더를 생성하는 방법을 찾아봐야겠습니다.

현재는 이런식으로 camel 컨텍스트에 기본설정만 하고있어서 좀 더 공부를 해야겠습니다.

        <route>
            <from uri="netty4:tcp://127.0.0.1:8989?decoders=#customDataDecoder,#nettyProcessRouter"/>
            <log message="[TimeStamp :${date:hh:MM:ss.SSS}]" />
        </route>

[김남철]

디코더가 이렇게 복잡해질 이유는 없다고 생각됩니다.

제가 쓰는 방식의 디코더 구현을 대충 정리드리자면

protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

    if (in.readableBytes() < 헤더사이즈) {

        return;

    }

    in.markReaderIndex();

    //헤더 데이타를 ByteBuf에서 처리하고요.

    if (in.readableBytes() < 받아야할 데이타 사이즈)  {

        in.resetReaderIndex();

  return;

    }

    //이제 다 받았으므로 

    out.add();

}

이런식으로 처리하도록 단순화 해보시길...

--

이 메일은 Google 그룹스 'Netty Korean User Group' 그룹에 가입한 분들에게 전송되는 메시지입니다.
이 그룹에서 탈퇴하고 더 이상 이메일을 받지 않으려면 netty-ko+u...@googlegroups.com에 이메일을 보내세요.

웹에서 이 토론을 보려면 https://groups.google.com/d/msgid/netty-ko/e51ecbee-989a-4549-bba1-074594afd458%40googlegroups.com을(를) 방문하세요.

[Jeonghee Song]

미남님 답변 감사합니다~

역시 저의 디코더 형식은 영 이상한 파스타코드였군요..

디코더 인스턴스를 커넥션 별로 생성 시키고 디코더를 단순화 시키면 해결될것 같습니다. 빛이 보이네요 ^^

[Jeonghee Song]

이런 방식으로 해결 완료 하였습니다. 감사합니다^^



<route>
            <from uri="netty4:tcp://127.0.0.1:8989?decoder=#testSeverInitializer "/>

            <log message="[TimeStamp :${date:hh:MM:ss.SSS}]" />
 </route>

public class TestSeverInitializer extends ChannelInitializer<Channel>{
   
    @EndpointInject
    protected ProducerTemplate template;
   
    protected void initChannel(Channel ch) throws Exception {
       
        int maxlength = 16383;
        byte[] lastDelimiter = {(char)127};
        ByteBuf delimiter = Unpooled.wrappedBuffer(lastDelimiter);
 
        ChannelPipeline channelPipeline = ch.pipeline();
       
        channelPipeline.remove("handler");
       
        channelPipeline.addLast("Decoder", new DelimiterBasedFrameDecoder(maxlength, delimiter));
       
        channelPipeline.addLast("Custom-Handler", new LogicProcessRouter(template));
       
    }
   
}