9. STREAM¶

가이드 홈 | 이전: 8. Session과 Actor binding | 다음: 10. Location — 자동 연결과 Object 위치

다른 언어로 보기 — C#/.NET · C++ · Java · Kotlin · Node/TypeScript

이 장의 계약 소유 문서 — STREAM 서버 session이 동작을, 언어별 STREAM session 공개 계약이 서버의 정확한 시그니처를 소유한다. Client package는 Stream Connector 가이드와 언어별 공개 계약을 따른다.

STREAM은 외부 client와 Framework server 사이의 연결 지향 양방향 메시지 채널이다. Server는 session lifecycle과 packet dispatch를 구현한다. Client는 독립 package인 Zlink.Stream.Connector를 사용한다.

1. Server node 등록¶

Stream node에는 session type 하나를 등록한다. Actor dispatch를 사용하면 명시적으로 활성화한다.

builder.addStreamNode('client-stream')
  .bind('tcp://0.0.0.0:9100')
  .enableActorDispatch()
  .registerSession(PlaySessionFactory); // 연결마다 만들 session factory를 등록한다.

Session handler와 Actor/Spot handler는 Framework의 기본 typed JSON serialization을 사용한다. Application이 message type마다 codec을 등록하거나 raw frame을 해석하지 않는다.

등록은 명시적이다. attribute·annotation·decorator로 stream node를 암시적으로 등록하는 표면은 없다. 축은 셋뿐이다 — node 이름, bind endpoint, session type. 그중 bind endpoint는 반드시 지정한다.

다음 여덟은 첫 연결까지 미루지 않고 host 시작 전에 설정 오류로 막는다.

조건
node 이름이 비어 있다
같은 node 이름을 두 번 등록했다
bind endpoint가 없다
같은 session type을 중복 등록했다
한 node에 session을 둘 이상 등록했다
TLS를 켰는데 인증서 경로가 비어 있다
TLS를 켰는데 key 경로가 비어 있다
TLS server를 설정하지 않고 client 인증서를 요구했다

TLS를 켜면 인증서와 key 경로를 함께 지정한다. client 인증서 요구는 기본이 꺼짐이고, 켜면 검증에 실패한 연결은 session을 만들기 전에 거부한다.

2. Session lifecycle¶

Session은 연결, packet dispatch, 오류와 disconnect callback을 구현한다. 같은 session의 callback은 직렬로 실행된다.

샘플에서 보기 — TicTacToe. client 연결 하나를 대표하는 session이다. 인증 packet을 먼저 거르고 나머지는 Actor로 relay한다. 저장소의 실제 코드다.

class PlaySession implements ZLinkSession {
  constructor(readonly context: ZLinkSessionContext) {}

  async onDispatch(dispatch: ZLinkSessionDispatchContext, payload: ZLinkMessage): Promise<void> {
    if (await this.context.handlers.tryHandle(dispatch, payload)) return;
    const actor = this.context.actors.bound.length === 1 ? this.context.actors.bound[0] : undefined;
    if (actor === undefined) throw new Error(`AuthenticateReq is required before '${dispatch.packetName}'.`);
    await actor.relay(payload);
  }

  async onDisconnected(): Promise<void> {
    // Framework cleanup notifies the exact bound-actor snapshot. Re-submitting
    // actor.notifyDisconnected() here can race a later session binding.
  }
}

최소 형태로 보면 이렇다.

export class PlaySession implements ZLinkSession {
  constructor(
    private readonly context: ZLinkSessionContext,
    private readonly logger: Logger
  ) {}

  configure(): void {
    // typed session packet handler를 등록한다.
    this.context.handlers.addHandler(PingHandler);
  }

  async onConnected(): Promise<void> {
    this.logger.log(`connected: ${this.context.sessionId}`);
  }

  async onDispatch(dispatch: ZLinkSessionDispatchContext, payload: ZLinkMessage): Promise<void> {
    if (await this.context.handlers.tryHandle(dispatch, payload)) return;
    // application protocol에 없는 packet을 받으면 연결을 닫는다.
    await this.context.close();
  }

  async onDisconnected(): Promise<void> {
    this.logger.log(`disconnected: ${this.context.sessionId}`);
  }
}

오류가 어디로 가는지는 넷으로 갈린다. session 오류 callback은 그 session에 귀속되는 transport 오류만 받는다.

오류	어디로 가나
그 session의 transport 오류	session 오류 callback
handshake 실패	runtime monitoring. session이 만들어지기 전이라 부를 대상이 없다
socket · node 단위 오류	runtime monitoring. session 하나의 오류로 확정할 수 없다
application handler 예외	handler 예외 처리 경로. session 오류 callback이 아니다

handler filter는 session dispatch에 적용되지 않는다. 다른 dispatch에 걸어 둔 filter가 있어도 session callback 앞에서는 돌지 않는다. 인증처럼 session 경로에서 걸러야 하는 일은 session의 handler 등록으로 처리한다.

recv loop를 직접 도는 표면은 없다. Framework가 packet을 queue에 넣은 뒤 session callback을 실행하며, 그 경계에서 dispatch · DI · logging을 일관되게 적용한다. loop · 취소 · backpressure를 application이 떠안지 않게 하려는 설계다.

3. Typed packet handler¶

Handler registry가 수신 message를 typed message로 decode한다. Request에 reply할 때는 현재 dispatch의 one-shot reply token을 사용한다.

async handle(
  context: ZLinkSessionContext, dispatch: ZLinkSessionDispatchContext, payload: ZLinkMessage) {
  if (!dispatch.canReply) throw new Error('Ping must be a request.');

  const message = payload.decode<Ping>(Object as never);
  // 같은 request correlation으로 한 번만 reply한다.
  await context.client.reply(pong(message.sequence)).submit();
}

reply는 현재 request에서만 유효하며 한 번 제출할 수 있다. Timeout이나 cancellation으로 전송이 실패해도 같은 reply token을 다시 사용할 수 없다.

응답에는 packet 이름이 실리지 않는다. client는 request sequence만으로 대기 중인 요청을 찾고, 응답을 어떤 타입으로 읽을지는 호출할 때 지정한 타입이 정한다. 이름으로 고르지 않으므로 응답 쪽에 packet 이름을 붙이는 표면도 없다. 오류 응답도 같은 sequence로 돌아온다.

Server가 먼저 push할 때는 send를 사용한다.

// local transport queue admission까지 기다린다.
await context.client
  .send(serverNotice('maintenance'))
  .metadata('severity', 'info')
  .compress()
  .submit();

4. Actor dispatch¶

인증 뒤 Actor를 session에 bind하고, session 전용 handler가 처리하지 않은 message를 session actor의 relay 호출로 넘길 수 있다. 상세 흐름은 Session과 Actor binding을 따른다.

Application은 session route를 Location Store에서 직접 조회하지 않는다. Actor relocation이 완료되면 Framework가 binding route를 갱신한다.

5. Client 연결¶

Client는 server Framework package가 아니라 Stream Connector package를 사용한다.

const client = zlinkStreamConnectorFactory.create({
  endpoint: 'tcp://game.example.com:9100',
  dispatchMode: ZlinkStreamDispatchMode.Manual
});

client.on(GameStateNotify, (message) => {
  render(message.payload);
});

await client.connect(); // 연결과 receive loop 준비를 완료한다.

while (running) {
  await client.dispatch(); // Manual 모드는 이 caller에서 callback을 실행한다.
}

게임 loop나 UI thread에서 callback을 실행해야 하면 Manual을 사용한다. Immediate는 connector의 worker에서 callback을 실행하므로 thread affinity가 필요한 client에는 적합하지 않다.

5.1 Diagnostics level¶

Connector는 server runtime과 같은 네 값(off/errors/Normal/Detailed)의 diagnostics level 옵션을 받는다. 기본값은 errors로 기존 동작과 같고, off로 낮추면 connector가 outbound frame에 flow 식별자를 만들거나 부착하지 않아 관측 전용 비용이 사라진다 (Stream Connector 공통 스펙 §13). Request/response 매칭에 쓰는 correlation은 protocol 정보라 off에서도 그대로 동작한다.

const options = {
    endpoint: 'ws://game.example.com:9100',
    diagnosticsLevel: ZlinkStreamDiagnosticsLevel.Off,
};

6. Client send와 request¶

// bounded outbound queue admission까지 기다린다.
await client.send(playerInput(direction)).submit();

// request sequence로 response를 찾는다.
const profile = await client.request(getProfile(playerId)).submit<Profile>();

Connector의 기본 typed codec은 JSON이다. Packet name override, push 대기, reconnect, heartbeat와 bounded queue 설정은 Stream Connector 가이드에서 설명한다.

7. 관련 문서¶

이 챕터 계약의 실행 검증 예문: 13. Interface 카탈로그 §5 — 검증 클래스 StreamContracts
Session과 Actor binding: Session Actor Dispatch
Client connector 전체 사용법: Stream Connector 가이드
Location Store와 자동 연결: Location