RSocket — 리액티브 메시징 프로토콜

┌──────────────────────────────────────────────┐
│              RSocket 프로토콜 스택              │
├──────────────────────────────────────────────┤
│  애플리케이션 계층: 4가지 상호작용 모델           │
│  (Request-Response, Fire-and-Forget,          │
│   Request-Stream, Channel)                    │
├──────────────────────────────────────────────┤
│  프레이밍 계층: 바이너리 프레임 인코딩/디코딩      │
│  (SETUP, REQUEST_RESPONSE, REQUEST_STREAM,    │
│   REQUEST_N, CANCEL, PAYLOAD, ERROR ...)      │
├──────────────────────────────────────────────┤
│  전송 계층: TCP / WebSocket / Aeron            │
│  (Netty가 이 계층을 담당)                       │
└──────────────────────────────────────────────┘

// 클라이언트 — 요청을 보내고 단일 응답 수신
Mono<Payload> response = rSocket.requestResponse(
    DefaultPayload.create("Hello")  // 요청 페이로드
);

response.subscribe(payload ->
    System.out.println("응답: " + payload.getDataUtf8())
);

// 메트릭 전송 — 응답이 필요 없는 시나리오
Mono<Void> result = rSocket.fireAndForget(
    DefaultPayload.create("{\"cpu\": 72.5, \"memory\": 4096}")
);

result.subscribe();  // 보내기만 하고 끝

// 주식 시세 구독 — 하나의 요청으로 지속적인 데이터 수신
Flux<Payload> stream = rSocket.requestStream(
    DefaultPayload.create("AAPL")  // 애플 주식 시세 요청
);

stream
    .doOnNext(payload ->
        System.out.println("시세: " + payload.getDataUtf8()))
    .take(100)       // 100개만 받겠다는 의미 — 배압 제어
    .subscribe();

// 양방향 채팅 — 클라이언트와 서버가 동시에 메시지 스트림을 교환
Flux<Payload> clientMessages = Flux.interval(Duration.ofSeconds(1))
    .map(i -> DefaultPayload.create("클라이언트 메시지 #" + i));

Flux<Payload> serverResponses = rSocket.requestChannel(clientMessages);

serverResponses
    .doOnNext(payload ->
        System.out.println("서버로부터: " + payload.getDataUtf8()))
    .subscribe();

수신자 → 송신자: REQUEST_N(n=5)    // "5개만 보내줘"
송신자 → 수신자: PAYLOAD (1)
송신자 → 수신자: PAYLOAD (2)
송신자 → 수신자: PAYLOAD (3)
수신자 → 송신자: REQUEST_N(n=3)    // "3개 더 보내도 돼"
송신자 → 수신자: PAYLOAD (4)
송신자 → 수신자: PAYLOAD (5)
송신자 → 수신자: PAYLOAD (6)

1. 최초 연결: SETUP 프레임에 Resume Token 포함
2. 연결 끊김 감지
3. 새 TCP 연결 수립
4. RESUME 프레임 전송 (Resume Token + 마지막 수신 위치)
5. 서버가 Resume Token 검증 → 이전 세션 복구
6. 끊긴 지점부터 데이터 재전송

┌─────────────────────────────────────────────┐
│  RSocket API (Mono/Flux 기반 상호작용 모델)    │
├─────────────────────────────────────────────┤
│  RSocket-Core (프레임 생성, 스트림 관리)       │
├─────────────────────────────────────────────┤
│  RSocket-Transport-Netty                     │
│  ┌─────────────────────────────────────┐     │
│  │  Netty ChannelPipeline              │     │
│  │  ┌───────────┐  ┌────────────────┐  │     │
│  │  │ LengthField│  │ RSocket Frame │  │     │
│  │  │ Codec     │  │ Codec         │  │     │
│  │  └───────────┘  └────────────────┘  │     │
│  └─────────────────────────────────────┘     │
├─────────────────────────────────────────────┤
│  Netty EventLoop (NioEventLoopGroup)         │
├─────────────────────────────────────────────┤
│  TCP / WebSocket                             │
└─────────────────────────────────────────────┘

// RSocket 서버를 Netty TCP 전송으로 시작하는 코드
RSocketServer.create(
        SocketAcceptor.with(new MyRSocketHandler()))  // 요청 처리기
    .bind(TcpServerTransport.create("localhost", 7000))  // Netty TCP 전송
    .subscribe();

// WebSocket 전송으로 바꾸려면 전송 계층만 교체
RSocketServer.create(
        SocketAcceptor.with(new MyRSocketHandler()))
    .bind(WebsocketServerTransport.create("localhost", 8080))  // Netty WebSocket 전송
    .subscribe();

<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>

# application.yml
spring:
  rsocket:
    server:
      port: 7000              # RSocket 서버 포트
      transport: tcp           # tcp 또는 websocket

@Controller
public class GreetingController {

    // Request-Response: Mono 입력 → Mono 출력
    @MessageMapping("greeting")
    public Mono<String> greeting(String name) {
        return Mono.just("안녕하세요, " + name + "님!");
    }

    // Fire-and-Forget: 입력만 있고 반환은 Mono<Void>
    @MessageMapping("log")
    public Mono<Void> log(LogEvent event) {
        System.out.println("로그 수신: " + event);
        return Mono.empty();
    }

    // Request-Stream: 단일 입력 → Flux 출력
    @MessageMapping("stock.{symbol}")
    public Flux<StockPrice> stockStream(@DestinationVariable String symbol) {
        return stockService.getStream(symbol);  // 지속적으로 시세 전송
    }

    // Channel: Flux 입력 → Flux 출력
    @MessageMapping("chat")
    public Flux<ChatMessage> chat(Flux<ChatMessage> incoming) {
        return incoming
            .doOnNext(msg -> broadcast(msg))    // 받은 메시지 브로드캐스트
            .switchMap(msg -> messageStream());  // 다른 사용자 메시지 스트림 반환
    }
}

// RSocketRequester — Spring이 제공하는 클라이언트 API
RSocketRequester requester = RSocketRequester.builder()
    .tcp("localhost", 7000);

// Request-Response 호출
Mono<String> response = requester
    .route("greeting")           // route 메타데이터 지정
    .data("개발자")               // 요청 데이터
    .retrieveMono(String.class); // 단일 응답

// Request-Stream 호출
Flux<StockPrice> prices = requester
    .route("stock.AAPL")
    .retrieveFlux(StockPrice.class);  // 스트림 응답

클라이언트 ──RSocket──→ Gateway ──RSocket──→ 서비스 A
                           │
                           └──RSocket──→ 서비스 B