Session과 Actor 연결¶

이 장을 읽고 나면

외부 client의 연결 하나를 Actor 하나에 묶고, 그 Actor가 같은 연결로 알림을 보낼 수 있다. 이 장의 코드는 예제 저장소의 tutorial README에서 가져왔으며, README의 「내려받기와 설치」·「빌드」·「실행」 절을 따르면 아래 결과를 재현할 수 있다.

STREAM의 session은 연결이 끊기면 함께 끝난다. 플레이어의 상태는 그보다 오래 남아야 하고, 그 상태를 맡는 것이 Actor다. 이 장은 그 둘을 묶는다. 묶은 뒤에는 session이 다루지 않은 packet이 그 Actor로 간다. Actor는 같은 연결로 보낼 수 있다.

1. Actor와 session을 묶는 역할¶

Actor와 session을 묶으면 수명이 다른 연결과 Actor가 서로 이어진다. 같은 플레이어가 끊었다 다시 접속하면 연결은 새로 만들어지고 Actor는 이전 것이 그대로 남아 있다. 그래서 연결이 자기 Actor를 가리키게 해 두고, 그 뒤의 packet은 그 Actor가 받는다.

↗ 크게 보기

▶ 둘러보기에서 직접 눌러 보기 — 로비와 던전

묶은 뒤에도 session은 사라지지 않는다. session이 아는 packet은 session이 먼저 처리하고, 모르는 것만 Actor로 간다.

2. 주고받을 packet을 정한다¶

인증 요청과 그 답, 그리고 Actor가 보내는 알림이다. 마지막 것은 답할 요청이 없다.

data class Authenticate(val playerId: String)

data class Authenticated(val playerId: String)

// Pushed by the player to its own connection, with no request to answer.
data class NicknameChanged(val nickname: String)

3. 받는 쪽 — 연결을 Actor에 묶는 node¶

받는 쪽 process

이 절의 코드는 STREAM에서 stream node를 등록한 process에 들어간다. 등록할 때 actor dispatch를 켜 두어야 이 장의 relay가 동작한다.

3.1 묶기¶

client가 자기 id를 보내면, 그 id의 Actor를 찾거나 만들어 이 연결에 묶는다.

// Ties this connection to one player. After this, packets without a session
// handler reach that player, and the player can push to this connection.
class AuthenticateHandler(players: ZLinkActorManager) :
    ZLinkSuspendingTypedSessionPacketHandler<ZLinkSessionContext, Authenticate> {

    private val players = players.kotlin()

    override fun packetName(): String = "Authenticate"

    override fun messageType(): Class<Authenticate> = Authenticate::class.java

    override suspend fun handle(
        context: ZLinkSessionContext,
        dispatch: ZLinkSessionDispatchContext,
        message: Authenticate,
    ) {
        // A returning client finds its existing player rather than a new one.
        val result =
            players
                .getOrCreate(message.playerId, "player")
                .inMesh("game")
                .request(CreatePlayer(message.playerId))
                .timeout(Duration.ofSeconds(10))
                .await()

        val bound = context.actors().bindOrGetActor(resolve(result))

        context.client().kotlin().reply(Authenticated(bound.actorId())).await()
    }

    private fun resolve(result: ZLinkActorCreateResult): ActorRef =
        when (result) {
            is ZLinkActorCreateResult.Existing -> result.actor()
            is ZLinkActorCreateResult.Created -> result.actor()
            else -> throw IllegalStateException("Player creation was rejected.")
        }
}

같은 id로 다시 접속하면 있던 Actor에 묶인다. 그래서 만들기와 찾기를 한 호출로 처리한다.

3.2 남은 packet을 넘기기¶

session이 처리하지 못한 packet을 묶인 Actor로 보낸다. 대상은 packet마다 dispatch context의 Actor가 정한다 — client가 Actor handle로 보낸 packet은 그 Actor를 가리킨다. Actor handle 없이 보낸 packet은 Actor가 비어 있으므로, 이 연결에 묶인 Actor가 하나이면 그 Actor로 보낸다. 묶인 Actor가 없으면 보낼 곳이 없으므로 인증이 먼저라는 것이 여기서 강제된다. client 쪽 구분은 packet 수신 §8이 다룬다.

// A slotted packet uses its dispatch Actor. An unslotted packet can use
// the connection only while exactly one Actor is bound.
val player =
    dispatch.actor()
        ?: run {
            val bound = context.actors().bound()
            when (bound.size) {
                1 -> bound[0]
                0 -> error("Authenticate an Actor before sending player packets.")
                else -> error("Select an Actor handle when more than one Actor is bound.")
            }
        }

player.kotlin().relay(dispatch, payload).await()

넘어간 packet은 그 Actor의 handler가 받는다 — Actor에서 사용한 것과 같은 handler다. 같은 handler가 mesh 호출로도 STREAM relay로도 호출된다.

3.3 Actor가 그 연결로 보내기¶

Actor는 자기에게 묶인 연결을 알고 있다. 응답이 아니라 스스로 보내는 알림이다.

// Pushes over the connection bound to this player. The same handler also runs
// on an HTTP path with no bound connection, where push ends with InvalidOperation.
// Rename is already complete, so only that failure is discarded.
try {
    actor.context().boundSession().kotlin().send(NicknameChanged(actor.nickname)).await()
} catch (error: ZLinkFrameworkException) {
    if (error.kind() != ZLinkFrameworkErrorKind.INVALID_OPERATION) {
        throw error
    }
}

묶인 연결이 없으면 push는 InvalidOperation으로 끝난다

같은 handler는 mesh 호출로도 불리며, 그때는 묶인 연결이 없다. 이 경우 push는 호출의 terminal에서 InvalidOperation으로 끝나며, 언어별 비동기 완료 수단으로 그 실패를 관찰한다. 위 코드는 rename을 이미 끝낸 뒤 push하므로 그 실패만 받아 버린다.

4. 접속하는 쪽 — mesh 밖의 client¶

접속하는 쪽 process

이 절의 코드는 STREAM에서 연결을 연 process에 들어간다. 앞 절과 짝을 이룬다.

인증하고, 알림을 받는다. 두 번째 호출은 답을 기다리지 않는 단방향인데도 뒤에 값이 온다 — 그것이 Actor가 보낸 알림이다.

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}") }
    }
// With one Actor bound, the connector can send without an Actor handle.
val authenticatedP1 =
    connector.request<Authenticated>(Authenticate("p1")).timeout(Duration.ofSeconds(5)).await()

println("bound player: ${authenticatedP1.playerId}")

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()

// A second Actor on the same connection calls for explicit handles.
val authenticatedP2 =
    connector.request<Authenticated>(Authenticate("p2")).timeout(Duration.ofSeconds(5)).await()
println("bound player: ${authenticatedP2.playerId}")

val playerP1 = requireNotNull(connector.actor(authenticatedP1.playerId))
val playerP2 = requireNotNull(connector.actor(authenticatedP2.playerId))
println("actor handle: ${playerP1.actorId}")
println("actor handle: ${playerP2.actorId}")

// Each callback receives only the push for its handle's Actor.
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)
    }

// 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)
    }

// Each handle sends to its own player over the same connection.
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()}")

5. 실행 결과¶

tutorial README의 「실행」 절을 따라 띄운 상태에서 Server를 실행한 상태에서 아래 명령으로 StreamClient를 시작하면, StreamClient가 보낸 bind request와 받은 Actor 알림은 StreamClient process의 stdout에 나오고 session·Actor handler 기록은 Server process의 stdout 또는 server.log에 나온다.

dotnet run --project StreamClient/StreamClient.csproj
# connected: True
# round trip: 56ms
# bound player: p1
# pushed: speedy

pushed 줄은 client가 요구하지 않은 응답이다. client는 이름을 바꾸라고 보냈을 뿐 답을 요구하지 않았다. 돌아온 것은 player가 스스로 보낸 알림이다. 그 사이에 packet은 session을 지나 Actor로 갔고, Actor가 같은 연결로 되돌려 보냈다.

6. 연결이 끊긴 뒤 — session은 끝나고 Actor는 남는다¶

session은 끝나고 Actor는 남는다. 같은 id로 다시 접속하면 그 Actor에 다시 묶인다.

끊김을 Actor에게 알리는 callback, 묶음이 다른 연결로 옮겨 갈 때의 규칙, 그리고 Actor가 다른 node로 이동하는 중의 처리는 Session 묶음의 동작 원리가 다룬다.

7. 관련 문서¶

연결을 받는 자리 — STREAM
묶이는 대상 — Actor
묶음의 규칙 — Session 묶음의 동작 원리
이 장 코드의 실행본 — 예제 저장소의 tutorial README