springboot + tcp통신

외부에서 tcp 통신으로 메세지 혹은 전화번호 같은 정보를 springboot로 받았을 경우를 구현하였습니다.

tcp통신을 바이트로 변환해서 받았을 경우에 springboot에서는 바이트 코드를 구분지을 수 있는 방법을 선택해야합니다.
1. STX와 ETX로 바이트의 시작점과 끝을 구분하는 경우 (발송부분에서도 STX와 ETX를 구분지어서 발송)
2. 바이트의 길이로 구분을 짓는 방법

이 중에서 저는 바이트의 길이로 구분을 짓는 방법을 선택하였습니다.

의존성 추가

	implementation 'org.springframework.boot:spring-boot-starter-integration'
    implementation 'org.springframework.integration:spring-integration-ip:6.2.1'

서버 포트 작성

application.yml에 임시 포트를 작성하였습니다.

tcp:
  server:
    port: 13000

CustomTcpSerializer

TCP 직렬화 및 역직렬화를 담당하는 클래스 작성

public class CustomTcpSerializer extends AbstractPooledBufferByteArraySerializer {

    @Override
    public byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
        int length = inputStream.read();
        if (length < 0) {
            throw new SoftEndOfStreamException("Stream closed between payloads");
        }
        int totalLength = length;

        int n = 0;
        try {
            while (totalLength > n) {
                int bite = inputStream.read();
                checkClosure(bite);
                buffer[n++] = (byte) (bite);
                if (n >= getMaxMessageSize()) {
                    throw new IOException("Message too long for buffer: " + getMaxMessageSize());
                }
            }
            return copyToSizedArray(buffer, n);
        } catch (IOException e) {
            publishEvent(e, buffer, n);
            throw e;
        } catch (RuntimeException e) {
            publishEvent(e, buffer, n);
            throw e;
        }
    }

    @Override
    public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
        outputStream.write(bytes.length); // 길이를 먼저 씀
        outputStream.write(bytes);
        outputStream.flush(); // 강제로 출력 스트림 비우기
    }
}

여기서 outputStream.write(bytes.length)이런식으로 길이를 먼저 사용하지 않으면 메세지, 혹은 전화번호가 잘려서 전송이 되는 문제가 생길 수 있습니다.

TcpServerConfig

TCP 서버 설정 클래스 생성

@Configuration
@EnableIntegration
public class TcpServerConfig {
    private static final Logger logger = LoggerFactory.getLogger(TcpServerConfig.class);

    @Value("${tcp.server.port}")
    private int port;

    @Bean
    public CustomTcpSerializer serializer() {
        return new CustomTcpSerializer();
    }

    @Bean
    public AbstractServerConnectionFactory connectionFactory(CustomTcpSerializer serializer) {
        TcpNioServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(port);
        connectionFactory.setSerializer(serializer);
        connectionFactory.setDeserializer(serializer);
        connectionFactory.setSingleUse(true);

        return connectionFactory;
    }

    @Bean
    public MessageChannel inboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel replyChannel() {
        return new DirectChannel();
    }

    @Bean
    public TcpInboundGateway inboundGateway(AbstractServerConnectionFactory connectionFactory,
                                            @Qualifier("inboundChannel") MessageChannel inboundChannel,
                                            @Qualifier("replyChannel") MessageChannel replyChannel) {
        TcpInboundGateway tcpInboundGateway = new TcpInboundGateway();
        tcpInboundGateway.setConnectionFactory(connectionFactory);
        tcpInboundGateway.setRequestChannel(inboundChannel);
        tcpInboundGateway.setReplyChannel(replyChannel);
        return tcpInboundGateway;
    }
}

아래의 어노테이션을 통해 Spring Integration 설정 파일을 bean으로 등록합니다.
@Configuration : 설정 파일을 bean으로 등록하기 위한 어노테이션
@EnableIntegration: Spring Integration의 기능을 활성화하는 어노테이션

connectionFactory 메서드는 TcpNioServerConnectionFactory를 생성하고 구성합니다. 이 클래스는 TCP 서버와의 연결을 설정합니다.
setSerializer, setDeserializer에 CustomTcpSerializer 클래스를 인자로 넘겨서 직렬화 및 역직렬화에 사용하도록 설정합니다.
setSingleUse(true)는 서버가 한 번의 메시지를 처리한 후에 연결을 닫아야 함을 나타냅니다.

inboundChannel 및 replyChannel 메서드는 각각 DirectChannel 빈을 정의합니다. DirectChannel는 서로 다른 컴포넌트 간 통신에 사용되는 메시지 채널입니다.

inboundGateway 메서드는 TcpInboundGateway를 구성합니다. 이는 TCP 클라이언트로부터 수신된 메시지를 처리하는 데 사용됩니다. 해당 메서드에서는 연결 팩토리, 인바운드 및 리플라이 채널을 설정합니다.
setConnectionFactory 함수의 인자로는 connectionFactory 함수에서 생성한 TcpNioServerConnectionFactory 객체를 넘겨줍니다.
setRequestChannel, setReplyChannel 함수의 인자로는 inboundChannel, replyChannel 함수에서 생성한 DirectChannel 객체를 넘겨줍니다.

@Qualifier("inboundChannel")로 명시해주지 않으면 springboot가 많은 bean타입으로 인하여 어떤 bean타입을 참조해야 하는지 구분짓지 못하는 오류가 터질 수 있음.

TcpServerEndpoint

TCP 서버의 엔드포인트를 정의하는 클래스 작성

@MessageEndpoint
public class TcpServerEndpoint {
    private final TcpMessageService tcpMessageService;

    public TcpServerEndpoint(TcpMessageService messageService) {
        this.tcpMessageService = messageService;
    }

    @ServiceActivator(inputChannel = "inboundChannel", async = "true")
    public void process(byte[] message) {
        tcpMessageService.processMessage(message);
//        System.out.println("Processed message: " + new String(message));
    }
}

@MessageEndpoint 어노테이션은 해당 클래스가 메시지 엔드포인트임을 나타내는 어노테이션입니다. 메시지 엔드포인트는 메시지를 수신하고 처리하는 역할을 합니다.

@ServiceActivator 어노테이션은 메서드가 특정 채널(inboundChannel)에서 메시지를 수신하여 활성화되도록 지정합니다. async = "true" 옵션은 이 메서드가 비동기적으로 동작함을 나타냅니다.
process 메서드는 inboundChannel 채널에서 들어오는 메시지를 처리하는데 사용됩니다.
메시지는 바이트 배열로 전달되며, 이를 messageService를 통해 처리한 뒤 그 결과를 반환합니다.

TcpMessageService

메세지를 처리할 클래스 작성

@Service
public class TcpMessageService {
    private static final Logger logger = LoggerFactory.getLogger(TcpMessageService.class);
    private final ExecutorService executorService;
    private final FCMNotificationService fcmNotificationService;

    public TcpMessageService(FCMNotificationService fcmNotificationService) {
        this.fcmNotificationService = fcmNotificationService;
        this.executorService = Executors.newFixedThreadPool(10);
    }

    public void processMessage(byte[] message) {
        executorService.submit(() -> {
            String receivedMessage = new String(message);
            String phoneNumber = receivedMessage.replaceAll("-", "").trim();
            if (phoneNumber.startsWith("010")) {
                phoneNumber = phoneNumber.replaceAll("010", "+8210");
            }
            logger.info("Received message: {}", phoneNumber);
            System.out.println("Received message: " + phoneNumber);

            fcmNotificationService.sendNotification(phoneNumber);

        });
    }
}

TcpMessageService 함수에서 전달 받은 메세지를 처리하는 로직을 작성해줍니다.

그리고 서버를 실행하고 TCP 클라이언트에서 메세지가 송신되면 processMessage 함수에서 메세지가 수신되는것을 확인할 수 있습니다.

저는 tcp통신으로 전화번호 데이터를 받아 flutter 어플리케이션으로 전화번호를 발송하는 구문을 작성하였습니다.
fcmNotification을 이용하였습니다.

TcpServerApplicationTests

테스트 코드 작성

@SpringBootTest(classes = Main.class)
@ActiveProfiles("test")
public class TcpServerApplicationTests {

    @Value("${tcp.server.port}")
    private int port;

    @Test
    public void testTcpServer() throws Exception {
        try (Socket socket = new Socket("localhost", port);
             OutputStream outputStream = socket.getOutputStream();
             InputStream inputStream = socket.getInputStream()) {

            String phoneNumber = "010-1234-5678!";
            byte[] phoneNumberBytes = phoneNumber.getBytes();
            byte[] lengthBytes = new byte[]{(byte) phoneNumberBytes.length};

            // 메시지를 길이, 내용 형식으로 작성
            outputStream.write(lengthBytes); // Length
            outputStream.write(phoneNumberBytes); // Content
            outputStream.flush();

            // 응답 수신
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
            String response = reader.readLine();
            assertEquals("전화번호를 수신했습니다: " + phoneNumber, response);
        }
    }
}

보내려는 정보를 바이트로 변환시켜서, 바이트의 길이를 명시해서 보내는 방식으로 테스트를 작성하였다.
정보를 길이로 구분짓기 위해서는 발송하는 쪽에서도 outputStream.write(lengthBytes);메세지를 길이 형식을 반드시 작성 해줘야 한다.
STX와 ETX방식과는 다를 수 있으니 유의..!

참고 사이트 (https://syk531.tistory.com/63)