Skip to content

05-RSocket와-TCP.md

Latest commit

 

History

History
161 lines (118 loc) · 10.9 KB

05-RSocket와-TCP.md

File metadata and controls

161 lines (118 loc) · 10.9 KB
Raw

05. RSocket 통합과 TCP 전송

1. 무엇을 / 왜

RSocket은 리액티브 스트림 의미론을 가진 양방향 메시징 프로토콜로, request/response·fire-and-forget·request/stream·channel 네 가지 상호작용을 지원한다. rsocket 패키지는 챕터 03의 메서드 핸들러 엔진과 챕터 02의 변환 계층을 RSocket에 연결해, @MessageMapping 메서드로 RSocket 요청에 응답하고 RSocketRequester로 요청을 보내며, 나아가 @RSocketExchange 인터페이스를 선언형 클라이언트 프록시로 만든다.

tcp 패키지는 별개의 작은 추상화로, STOMP 브로커 릴레이(챕터 04)가 외부 브로커와 통신할 때 쓰는 TCP 연결 계약을 정의하고 Reactor Netty로 구현한다. RSocket과 직접 관계는 없지만, 둘 다 "리액티브 네트워크 전송"이라는 점에서 이 챕터에 함께 둔다.

서브패키지:

  • rsocket — RSocketRequester(클라이언트 API), RSocketStrategies(인코더·디코더·메타데이터 묶음), 메타데이터 인코딩/추출.
  • rsocket/annotation + support — @ConnectMapping, RSocketMessageHandler(서버/응답자).
  • rsocket/service — @RSocketExchange 인터페이스 → 클라이언트 프록시(RSocketServiceProxyFactory).
  • tcp + tcp/reactor — TcpOperations 계약과 Reactor Netty 구현.

2. 핵심 타입

  요청 보내기 (클라이언트)
  ──────────────────────
  RSocketRequester                          ── RSocket 위 플루언트 API
     route("route.path", vars)              → RequestSpec
        .metadata(...)                      → 라우팅/인증 등 메타데이터
        .data(payload)                      → RetrieveSpec
            .retrieveMono(Type) : Mono<T>   → request/response
            .retrieveFlux(Type) : Flux<T>   → request/stream
            .send()             : Mono<Void>→ fire-and-forget
     RSocketStrategies : Encoder/Decoder + MetadataExtractor + ReactiveAdapterRegistry

  요청 받기 (서버/응답자)
  ──────────────────────
  RSocketMessageHandler  extends MessageMappingMessageHandler (handler/annotation/reactive)
                         extends AbstractMethodMessageHandler (리액티브, 챕터 03)
     - @MessageMapping / @ConnectMapping 메서드를 RSocket 라우트로 매핑
     - RSocketFrameTypeMessageCondition 으로 상호작용 종류(FNF/RR/RS/RC) 매칭
     - responder() → io.rsocket.SocketAcceptor 어댑터

  선언형 클라이언트
  ────────────────
  @RSocketExchange 인터페이스
     └─ RSocketServiceProxyFactory.createClient(MyService.class) → 프록시
          (RSocketServiceArgumentResolver 가 @Payload/@DestinationVariable/메타데이터 해석)

RSocketRequester — 클라이언트 플루언트 API

rsocket/RSocketRequester는 보내는 RSocket을 감싼 얇은 래퍼다. 상호작용 종류는 호출하는 메서드가 결정한다 — retrieveMono면 request/response, retrieveFlux면 request/stream, send면 fire-and-forget.

// 사용 예 (개념)
requester.route("locate.radars.within", area)   // 라우트 메타데이터
         .data(fireRequest)                      // 페이로드(자동 인코딩)
         .retrieveFlux(AirportLocation.class);   // → request/stream

페이로드 인코딩/디코딩과 메타데이터 처리는 RSocketStrategies(인코더·디코더·MetadataExtractor·ReactiveAdapterRegistry 묶음)가 담당한다. DefaultRSocketRequester/DefaultRSocketRequesterBuilder가 구현이고, MetadataEncoder/DefaultMetadataExtractor가 라우트·인증 등 복합 메타데이터를 RSocket 메타데이터 포맷으로 변환한다.

RSocketMessageHandler — 서버 측 응답자

rsocket/annotation/support/RSocketMessageHandler는 챕터 03의 리액티브 메서드 핸들러(MessageMappingMessageHandler → handler/invocation/reactive/AbstractMethodMessageHandler)를 RSocket에 특화한 것이다.

  • @MessageMapping/@ConnectMapping/@RSocketExchange 메서드를 라우트로 매핑하고, 입력·출력 카디널리티(Mono vs Flux)로 RSocket 상호작용 종류를 자동 판정한다.
  • 매핑 조건에 RSocketFrameTypeMessageCondition을 추가해 프레임 타입(REQUEST_RESPONSE 등)까지 매칭한다.
  • responder()로 io.rsocket.SocketAcceptor를 얻어 RSocketServer에 등록하거나, RSocketRequester.Builder의 클라이언트 응답자로 쓴다(클라이언트도 서버로부터 요청받을 수 있는 RSocket 양방향성 지원).

@RSocketExchange — 선언형 클라이언트

rsocket/service는 spring-web의 HTTP 인터페이스와 동형의 패턴이다. 인터페이스에 @RSocketExchange를 선언하면 호출이 자동으로 RSocket 요청으로 변환된다.

interface RadarService {
    @RSocketExchange("locate.radars.within")
    Flux<AirportLocation> radars(@Payload MapRequest area);
}
RadarService svc = RSocketServiceProxyFactory.builder(requester).build()
                       .createClient(RadarService.class);

RSocketServiceProxyFactory가 인터페이스를 프록시로 만들고, RSocketServiceArgumentResolver 구현들(PayloadArgumentResolver·DestinationVariableArgumentResolver·MetadataArgumentResolver)이 메서드 인자를 RSocketRequestValues로 모아 RSocketRequester 호출로 변환한다. RSocketExchangeReflectiveProcessor/...AotProcessor는 AOT(네이티브 이미지) 힌트를 등록한다.

3. 동작 흐름

3.1 서버 측 요청 처리

RSocket 프레임 수신 (io.rsocket)
   │  RSocketMessageHandler.responder() 가 만든 SocketAcceptor 경유
   ▼
MessagingRSocket  ── 프레임을 Message 로 변환(라우트 메타데이터 → 목적지 헤더)
   │
   ▼
RSocketMessageHandler.handleMessage(message)    (리액티브 AbstractMethodMessageHandler)
   │  1. 라우트로 @MessageMapping 메서드 매칭 (+ 프레임 타입 조건)
   │  2. 리액티브 인자 리졸버로 @Payload 등 해석 (Decoder 사용)
   │  3. 메서드 호출 → 반환값(Mono/Flux)
   ▼
RSocketPayloadReturnValueHandler  ── 반환 Publisher 를 RSocket 응답 스트림으로 인코딩

핵심은 챕터 03 엔진을 그대로 쓰되, 인자 해석에 Decoder를, 반환 인코딩에 Encoder를 쓰는 리액티브 변형이라는 점이다. AbstractEncoderMethodReturnValueHandler가 반환 Publisher를 인코딩하고, ChannelSendOperator(handler/invocation/reactive)가 다운스트림 구독 타이밍을 조율한다.

3.2 클라이언트 측 요청 보내기

requester.route(...).data(payload).retrieveFlux(Type)
   │  MetadataEncoder : route/auth → RSocket 메타데이터
   │  Encoder         : payload    → RSocket 데이터 페이로드
   ▼
io.rsocket RSocket.requestStream(payload)
   │  응답 Flux<Payload>
   ▼
Decoder : Payload → Type    → Flux<Type>

@RSocketExchange 프록시를 쓰면 이 체인이 메서드 호출 한 번 뒤에 숨는다.

3.3 TCP 전송 (STOMP 브로커 릴레이용)

tcp/TcpOperations<P>는 "TCP 연결을 연다"는 계약이다.

// TcpOperations.java
CompletableFuture<Void> connectAsync(TcpConnectionHandler<P> handler);
CompletableFuture<Void> connectAsync(TcpConnectionHandler<P> handler, ReconnectStrategy strategy);
CompletableFuture<Void> shutdownAsync();
  • TcpConnectionHandler가 연결 수립/메시지 수신/끊김 이벤트 콜백을 받고, TcpConnection으로 메시지를 보낸다.
  • ReconnectStrategy(예: FixedIntervalReconnectStrategy)가 끊겼을 때 재연결 간격을 정한다 — 브로커 릴레이의 안정성 핵심.
  • tcp/reactor/ReactorNettyTcpClient가 Reactor Netty 기반 구현이며, ReactorNettyCodec(STOMP는 StompReactorNettyCodec/StompTcpMessageCodec)으로 바이트 ↔ Message 변환을 처리한다.

챕터 04의 StompBrokerRelayMessageHandler가 바로 이 TcpOperations로 외부 STOMP 브로커에 연결한다.

4. 핵심 메서드

  • RSocketRequester.route(...).data(...).retrieveMono/retrieveFlux/send: 상호작용 종류를 호출 형태로 표현하는 플루언트 종결 메서드. 반환 타입(Mono/Flux)과 종결 메서드가 RSocket 프레임 타입을 결정한다.
  • RSocketMessageHandler.responder(): 핸들러를 io.rsocket.SocketAcceptor로 어댑팅 — Spring 메시징 세계와 RSocket 라이브러리의 경계를 잇는 지점.
  • RSocketServiceProxyFactory.createClient(interfaceType): @RSocketExchange 인터페이스를 동적 프록시로 만들어, 메서드 호출을 인자 리졸버 체인을 거쳐 RSocketRequester 호출로 변환.
  • TcpOperations.connectAsync(handler, reconnectStrategy): 재연결 전략과 함께 연결을 열고, 끊김 시 자동 복구를 위임. 브로커 릴레이의 장애 내성 기반.

5. 설계 포인트 / 확장점

  • 메서드 핸들러 엔진의 두 번째 재사용: STOMP가 블로킹 엔진을 썼다면, RSocket은 같은 엔진의 리액티브 버전(handler/invocation/reactive)을 쓴다. 챕터 03의 추상화가 두 프로토콜·두 동시성 모델 모두를 커버함을 보여 준다.
  • 카디널리티 기반 상호작용 추론: 메서드 시그니처(Mono/Flux 입출력)만으로 RSocket 상호작용 종류를 자동 결정 — 개발자가 프레임 타입을 명시할 필요가 없다.
  • 선언형 클라이언트 패턴: @RSocketExchange + RSocketServiceProxyFactory는 spring-web의 HTTP 인터페이스, Spring Data 리포지토리와 같은 "인터페이스 선언 → 프록시 생성" 계열. 인자 리졸버 합성으로 확장한다.
  • 전송 추상화 분리: tcp는 STOMP 릴레이가 의존하는 최소 계약만 정의하고 구현(Reactor Netty)을 분리해, 다른 TCP 클라이언트로의 교체 여지를 남긴다.
  • 확장점: RSocketStrategies에 커스텀 Encoder/Decoder·MetadataExtractor 등록, RSocketConnectorConfigurer로 저수준 커넥터 조정, 커스텀 RSocketServiceArgumentResolver, 커스텀 ReconnectStrategy.

6. 정리

rsocket 패키지는 챕터 02의 변환 계층과 챕터 03의 리액티브 메서드 핸들러 엔진을 RSocket에 결합한다. RSocketRequester는 라우트·데이터·종결 메서드로 네 가지 상호작용을 플루언트하게 표현하고, RSocketMessageHandler는 @MessageMapping/@ConnectMapping을 라우트와 프레임 타입으로 매칭해 응답하며, @RSocketExchange + RSocketServiceProxyFactory는 인터페이스를 선언형 클라이언트로 바꾼다. tcp 패키지는 STOMP 브로커 릴레이가 외부 브로커와 통신하는 재연결 가능한 TCP 계약을 제공한다. 이로써 spring-messaging의 공통 추상화가 인메모리 채널·STOMP·RSocket이라는 세 가지 구체 세계를 모두 동일한 어휘로 다룬다는 그림이 완성된다.