packet 수신

목차 | 이전: packet 송신 | 다음: 연결 생명주기

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

이 장을 읽고 나면

서버가 먼저 보내는 packet을 handler로 받고, 등록을 원하는 시점에 해제할 수 있다. 특정 packet 하나를 기다리는 표면과 수신 개수를 읽는 표면도 함께 사용한다.

서버가 보낸 packet은 handler로 넘어가거나 대기 표면이 가져가기 전까지 수신 큐에 머문다. 이 장은 그 큐에서 packet을 꺼내는 방법을 다룬다. 응답과 heartbeat는 이 큐를 거치지 않는다 — 응답은 기다리는 request로 바로 이어지고, heartbeat는 연결 유지에 사용한다.

1. 타입 지정 수신과 디코딩

서버가 보낸 packet 이름으로 handler를 찾은 뒤, connector가 payload를 디코딩해 message에 담는다. message는 packet 이름, payload, metadata와 Actor ID를 함께 전달한다. bytes를 업무 타입으로 바꾸는 역할을 맡은 객체가 codec이다. codec은 connector 생성 옵션에 하나만 등록한다.

1.1 받는 쪽 — 타입을 지정한 handler

다음 코드는 tutorial의 수신 등록이다. .NET·Java·Kotlin·C++ 예제는 기존 JSON 흐름의 NicknameChanged를 받고, Node 예제는 Protobuf 흐름의 생성 클래스 Ping을 받는다. 각 예제의 실행 절차는 해당 tutorial README가 설명한다.

↗ 크게 보기

val singleReceive =
    connector.on<NicknameChanged> { message ->
        singleChanged.complete(message)
        CompletableFuture.completedFuture(null)
    }

1.2 타입이 codec으로 전달되는 순서

on(Type, handler)는 타입에서 packet 이름을 정하고, 그 타입으로 payload를 디코딩한다. 이름을 명시하는 형태는 이름으로 packet을 고르면서 디코딩 타입을 별도로 지정한다. 타입의 이름과 서버 packet 이름이 다를 때 사용하는 형태다.

on<T> { ... }는 T::class.java를 Java connector에 전달한다. 이름을 지정하는 on(name, T::class, handler)도 같은 Java decoder를 사용한다.

Node Protobuf codec

수정한 codec은 on(Type, handler)의 생성자로 디코딩한다. 타입이 없으면 factory의 기본 타입을 쓴다. envelope codec도 handler 타입을 decoder에 전달한다. 배포된 0.28.0과의 차이와 실행 절차는 Node Protobuf 송수신이 설명한다.

2. 등록 해제

등록하면 해제할 수 있는 값이 돌아온다. 화면 하나의 수명에 맞춰 구독을 등록했다가 화면을 닫을 때 해제하는 client가 연결까지 다시 만들지 않게 하기 위해서다. 해제한 handler는 그 뒤로 실행되지 않고, 같은 값을 두 번 해제해도 오류가 아니다.

subscription.close()

돌려받은 값의 수명이 등록의 수명인지는 언어가 정한다. 소유권을 값으로 나타내는 언어에서는 그 값이 사라지면 등록도 함께 사라지므로, 등록이 살아 있어야 하는 동안 그 값을 보관한다. 나머지 언어에서는 값을 버려도 명시적으로 해제할 때까지 등록이 유지된다.

연결 상태·오류·끊김 handler도 같은 값을 돌려준다. 이 handler들은 연결 생명주기가 다룬다.

3. handler가 실행되는 시점

기본 설정에서 receive 경로는 handler를 직접 호출하지 않고 내부 큐에 넣는다. application이 pump를 호출하면 그때까지 쌓인 handler가 pump를 호출한 실행 문맥에서 실행된다. 게임 loop라면 frame마다 한 번 호출한다. pump는 쌓인 것을 처리하고 돌아오며, 새 packet이 도착하기를 기다리지 않는다.

즉시 실행으로 바꾸면 receive 경로에서 handler가 바로 실행되고 pump를 호출할 필요가 없다. 대신 느린 handler가 receive 경로를 막으므로 뒤따르는 수신 처리가 그만큼 늦어진다.

while (running) {
    connector.dispatch().await()
    renderFrame()
}

아래 대기 표면은 등록된 handler가 아니라 수신 큐를 직접 관측한다. 그래서 pump를 호출하지 않는 구성에서도 그대로 동작한다.

4. packet 하나 기다리기

시나리오의 특정 지점에서 packet 하나를 기다릴 때는 handler를 등록하지 않고 대기 표면을 사용한다. 조건에 맞는 packet을 소비하고 그 message를 돌려주며, 조건에 맞지 않는 packet은 큐에 남아 이후의 handler나 대기가 처리한다. timeout을 지정하지 않으면 connector의 기본 대기 timeout을 사용한다.

packet 이름은 명시하거나 payload 타입에서 정할 수 있다.

waitFor<T>()와 waitFor<T>(name)을 사용한다.

다음 예제는 payload 타입으로 packet 하나를 기다린다.

val found =
    connector
        .waitFor<MatchFound>()
        .where { it.payload.matchId == "match-7f3a" }
        .timeout(Duration.ofSeconds(30))
        .await()

술어와 반환은 payload가 아니라 message를 다룬다. payload만 받으면 술어가 packet 이름과 metadata를 보지 못하기 때문이다. 걸러야 하는 값이 payload 안에 있으면 술어 안에서 그 필드를 읽는다 — 상태 값 전용 표면은 두지 않는다.

5. 오지 않아야 하는 packet과 도착 순서

시나리오를 검증할 때는 "왔다"뿐 아니라 "오지 않았다"와 "순서대로 왔다"도 확인한다. 앞은 관찰 구간을 지정하고 그 구간 동안 그 packet이 오지 않는지 보고, 뒤는 술어를 순서대로 적용해 같은 이름의 packet이 그 순서로 도착했는지 확인한 뒤 message 목록을 돌려준다.

connector.expectNone<OrderChanged>().within(Duration.ofMillis(100)).await()
connector.send(ReceivingStage("orders")).await()
val steps =
    connector
        .waitForSequence<OrderChanged>()
        .expect { it.payload.status == "paid" }
        .expect { it.payload.status == "shipped" }
        .timeout(Duration.ofSeconds(2))
        .await()

관찰 조건이 어긋난 실패 — 시간 안에 오지 않음, 오지 않아야 할 것이 도착함, 순서가 어긋남 — 는 검증 실패로 전달한다. 연결이 끝나 대기를 이어갈 수 없으면 연결 없음으로 전달한다. 관측이 어긋난 것과 관측할 자리가 사라진 것을 호출자가 구분해야 하기 때문이다.

6. 수신 개수

packet 이름별로 받은 개수를 읽는다. 소비해도 값이 줄지 않으므로, handler를 걸어 두고 pump한 뒤에도 그 이름으로 몇 건이 왔는지 판정할 수 있다. handler 실행 시점 설정과도 무관하게 packet이 도착한 시점에 값이 올라간다.

val count = connector.receivedCount("LeaderboardUpdate")

기준점은 연결이 성립한 시점이다. 연결이 성립하면 0에서 시작하고, 재연결하면 새 연결이므로 다시 0에서 시작한다. 이때 이전 연결에서 남아 있던 미소비 message도 함께 비운다 — 그러지 않으면 대기 표면이 끊기기 전의 packet을 새 연결의 것으로 돌려준다.

7. 수신 큐 — 한도를 두지 않는다

connector는 받은 것을 계속 받아서 처리한다. 큐에 한도를 두지 않고, message를 버리지 않으며, 큐가 길어졌다는 이유로 연결을 닫지도 않는다. connector는 socket을 직접 구현하지 않고 실행 환경이 주는 것을 사용하므로 읽기를 보류할 자리가 없다.

정상 동작하는 client에서는 message가 쌓이지 않는다. handler가 처리하거나 대기 표면이 소비하기 때문이다. 계속 쌓인다면 pump를 호출하지 않는 client 쪽 문제이므로, 수신 개수를 흐름 제어의 근거로 사용하지 않는다.

8. Actor가 여럿일 때 — Actor handle로 구분한다

8.1 Actor가 하나일 때

서버가 이 연결에 Actor를 하나만 bind하면 지금까지의 송수신 코드를 그대로 쓴다. connector에서 보낸 packet에는 Actor slot이 없고, 서버 session은 slot이 없는 packet을 그 연결에 묶인 하나의 Actor로 넘긴다(Session과 Actor 연결).

val singleChanged = CompletableFuture<ZLinkStreamMessage<NicknameChanged>>()
val singleReceive =
    connector.on<NicknameChanged> { message ->
        singleChanged.complete(message)
        CompletableFuture.completedFuture(null)
    }
connector.send(ChangeNickname("speedy")).await()
val pushed = singleChanged.await()
println("pushed: ${pushed.payload().nickname}, actor: ${pushed.actorId()}")
singleReceive.close()

8.2 bound 통지로 handle을 받는다

서버가 한 연결에 Actor를 여럿 bind하면, connector는 Actor마다 Actor handle을 만든다. 서버가 Actor를 bind하면 bound 통지가 그 Actor의 첫 packet보다 먼저 오고, bind가 끝나면 unbound 통지가 그 Actor의 마지막 packet 뒤에 온다. 다음 예제는 두 통지를 등록하고, 인증한 p1·p2의 handle을 Actor ID로 찾는다.

val boundNotice =
    launch(start = CoroutineStart.UNDISPATCHED) {
        connector.actorBound().collect { actor -> println("actor bound: ${actor.actorId}") }
    }
val unboundNotice =
    launch(start = CoroutineStart.UNDISPATCHED) {
        connector.actorUnbound().collect { actor -> println("actor unbound: ${actor.actorId}") }
    }
val playerP1 = requireNotNull(connector.actor(authenticatedP1.playerId))
val playerP2 = requireNotNull(connector.actor(authenticatedP2.playerId))
println("actor handle: ${playerP1.actorId}")
println("actor handle: ${playerP2.actorId}")

8.3 handle로 보내고 handle별로 받는다

handle로 보낸 packet에는 그 Actor의 slot이 실리고, 서버 session은 그 slot의 Actor로 packet을 넘긴다. handle에 등록한 수신 handler는 그 Actor가 보낸 message만 받는다.

val changedP1 = CompletableFuture<ZLinkStreamMessage<NicknameChanged>>()
val changedP2 = CompletableFuture<ZLinkStreamMessage<NicknameChanged>>()
val receiveP1 =
    playerP1.on<NicknameChanged> { message ->
        changedP1.complete(message)
        CompletableFuture.completedFuture(null)
    }
val receiveP2 =
    playerP2.on<NicknameChanged> { message ->
        changedP2.complete(message)
        CompletableFuture.completedFuture(null)
    }
playerP1.send(ChangeNickname("speedy-p1")).await()
playerP2.send(ChangeNickname("speedy-p2")).await()
val pushedP1 = changedP1.await()
val pushedP2 = changedP2.await()
println("pushed: ${pushedP1.payload().nickname}, actor: ${pushedP1.actorId()}")
println("pushed: ${pushedP2.payload().nickname}, actor: ${pushedP2.actorId()}")

실행하면 pushed: speedy-p1, actor: p1과 pushed: speedy-p2, actor: p2가 출력된다.

8.4 connector 수준에서 Actor ID로 가른다

handle 없이 connector 수준에서 받을 수도 있다. 이때 message의 Actor ID가 서버 쪽 상대 Actor를 알려 준다. slot 없이 온 message는 Actor ID가 비어 있다.

// Connector-level callbacks can distinguish the same pushes by ActorId.
val receiveActorIds =
    connector.on<NicknameChanged> { message ->
        println("received actor id: ${message.actorId()}")
        CompletableFuture.completedFuture(null)
    }

8.5 unbind된 Actor로 보낸 packet

unbind된 handle은 닫힌다. 닫힌 handle로 보내면 connector가 보내지 않고 validationFailed로 끝낸다. unbind 직전에 이미 보낸 packet이 서버에 늦게 도착하면 서버는 그 packet을 다른 Actor로 넘기지 않는다 — request는 InvalidOperation 오류 응답으로 끝나고, 단방향 send는 버려진다.

9. tutorial 실행 확인

등록 해제·pump·조건 대기·순서·수신 개수 예제는 각 언어 StreamClient의 --receiving 모드에 있다. STREAM_RECEIVING_ENDPOINT는 JSON packet을 보내는 서버 주소다. Node는 WebSocket 주소를, 나머지 예제는 TCP 주소를 사용한다. 검증용 peer는 LeaderboardUpdate를 두 번 보내고 MatchFound, OrderChanged를 정해진 순서로 보낸다. 다섯 언어에서 실제로 확인한 결과는 같다.

receiving: handler=1, frames=1, match=match-7f3a, sequence=paid,shipped, count=2

handler는 pump에서 한 번 실행되고 해제 뒤에는 실행되지 않는다. 두 번째 packet도 수신 개수에는 포함된다. 조건에 맞는 match와 paid·shipped 순서를 확인하며, 순서 검증 전에는 OrderChanged가 오지 않는지 확인한다.

10. 다음 장

• 연결 상태와 재연결, 종료 사유 — 연결 생명주기
• 수신 경로에서 나는 오류 — 오류 처리