Get back into a game a locked phone dropped out of

A phone locked when its turn came round could not play once it was
unlocked. The game was never wrong — the host still had the position and
the seat was still the player's — but the only route to it was gone, and
nothing rebuilt it. The board came back with the state it had before the
turn arrived, and every tap on it went nowhere.

A locked phone puts its Wi-Fi radio into power save, and a link carrying
nothing between one turn and the next is exactly what that drops. Four
things were missing, and all four are needed:

The radio is held awake for as long as there is a game on, which stops
most of these drops happening at all. FULL_HIGH_PERF rather than the
LOW_LATENCY its deprecation notice points at: that one only applies with
the screen on and the app in front, which is the case that worked.

A link with nothing on it is pinged, and answered. TCP does not report a
link that died this way — the read never returns and the next write goes
into a socket that leads nowhere — so without traffic of its own the app
could not tell a player who was thinking from one who was gone. Both ends
have to speak it or a quiet link would be dropped as dead, hence the
protocol bump; the lobby already turns a mismatch away with "update both
devices".

A write that fails now reports the link as broken instead of closing it as
though this end had meant to. That silence was the reason a dead board
answered nothing and said nothing about why.

And a guest that loses its link dials the host again — which is still
listening, on the same port, for the whole match — and is put back in the
seat it never actually left, by the device id it is already known by. The
table then resyncs, which is what fills the screen back in. Coming back
to the front asks for the same thing, since a client draws nothing but
what it was sent and anything pushed while the link was down was pushed
at nothing.

Sends are serialised while I am here: a message is a line, a write is
three calls, and the heartbeat writes on its own schedule.

Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GarwvAPvWHELNspro4foFY
This commit is contained in:
Claude
2026-08-16 00:28:05 +00:00
parent 32ab075ae2
commit f0c8895747
9 changed files with 806 additions and 72 deletions
+5
View File
@@ -6,6 +6,11 @@
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" /> <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<!-- Network Service Discovery needs multicast to see hosts on the LAN. --> <!-- Network Service Discovery needs multicast to see hosts on the LAN. -->
<uses-permission android:name="android.permission.CHANGE_WIFI_MULTICAST_STATE" /> <uses-permission android:name="android.permission.CHANGE_WIFI_MULTICAST_STATE" />
<!-- Holding a Wi-Fi lock keeps the radio out of power save while a game is
on, so a phone locked between turns does not drop out of it. The lock
is a wake lock as far as the system is concerned, hence the permission;
nothing here ever holds the screen or the CPU awake. -->
<uses-permission android:name="android.permission.WAKE_LOCK" />
<!-- Lets the app hand a downloaded APK to the system installer. --> <!-- Lets the app hand a downloaded APK to the system installer. -->
<uses-permission android:name="android.permission.REQUEST_INSTALL_PACKAGES" /> <uses-permission android:name="android.permission.REQUEST_INSTALL_PACKAGES" />
@@ -3,6 +3,7 @@ package org.prolibertate.games.net
import android.content.Context import android.content.Context
import android.net.nsd.NsdManager import android.net.nsd.NsdManager
import android.net.nsd.NsdServiceInfo import android.net.nsd.NsdServiceInfo
import android.net.wifi.WifiManager
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.awaitClose import kotlinx.coroutines.channels.awaitClose
@@ -31,17 +32,49 @@ class LanTransport(private val context: Context) : Transport {
private val nsdManager: NsdManager private val nsdManager: NsdManager
get() = context.getSystemService(Context.NSD_SERVICE) as NsdManager get() = context.getSystemService(Context.NSD_SERVICE) as NsdManager
private val wifiManager: WifiManager
get() = context.applicationContext.getSystemService(Context.WIFI_SERVICE) as WifiManager
private var serverSocket: ServerSocket? = null private var serverSocket: ServerSocket? = null
private var registrationListener: NsdManager.RegistrationListener? = null private var registrationListener: NsdManager.RegistrationListener? = null
private var discoveryListener: NsdManager.DiscoveryListener? = null private var discoveryListener: NsdManager.DiscoveryListener? = null
private var wifiLock: WifiManager.WifiLock? = null
override fun isAvailable(): Boolean = true override fun isAvailable(): Boolean = true
/**
* Keeps the Wi-Fi radio properly awake for as long as there is a game on.
*
* Android puts the radio into power save when the screen goes off, which is
* what a locked phone is, and a link left to it drops out from under a game
* that is still being played. FULL_HIGH_PERF is the mode that says otherwise
* — the deprecation notice points at LOW_LATENCY, which is the wrong one
* here: that one only applies while the screen is on and the app is in
* front, which is precisely the case that was never broken.
*/
@Suppress("DEPRECATION")
private fun holdRadio() {
if (wifiLock?.isHeld == true) return
val lock = wifiLock ?: runCatching {
wifiManager.createWifiLock(WifiManager.WIFI_MODE_FULL_HIGH_PERF, WIFI_LOCK_TAG)
.also { it.setReferenceCounted(false) }
}.getOrNull() ?: return
wifiLock = lock
runCatching { lock.acquire() }
}
private fun releaseRadio() {
wifiLock?.let { lock -> runCatching { if (lock.isHeld) lock.release() } }
wifiLock = null
}
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
// Hosting // Hosting
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
override fun host(displayName: String, scope: CoroutineScope): Flow<Connection> = callbackFlow { override fun host(displayName: String, scope: CoroutineScope): Flow<Connection> = callbackFlow {
holdRadio()
// Port 0 lets the OS pick a free port, which is then advertised. // Port 0 lets the OS pick a free port, which is then advertised.
val server = ServerSocket(0) val server = ServerSocket(0)
serverSocket = server serverSocket = server
@@ -66,6 +99,7 @@ class LanTransport(private val context: Context) : Transport {
while (isActive && !server.isClosed) { while (isActive && !server.isClosed) {
val socket = runCatching { server.accept() }.getOrNull() ?: break val socket = runCatching { server.accept() }.getOrNull() ?: break
socket.tcpNoDelay = true socket.tcpNoDelay = true
socket.keepAlive = true
val connection = StreamConnection( val connection = StreamConnection(
peerId = socket.inetAddress?.hostAddress ?: "unknown", peerId = socket.inetAddress?.hostAddress ?: "unknown",
kind = TransportKind.LAN, kind = TransportKind.LAN,
@@ -90,6 +124,8 @@ class LanTransport(private val context: Context) : Transport {
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
override fun discover(scope: CoroutineScope): Flow<List<DiscoveredHost>> = callbackFlow { override fun discover(scope: CoroutineScope): Flow<List<DiscoveredHost>> = callbackFlow {
holdRadio()
val found = Collections.synchronizedMap(linkedMapOf<String, DiscoveredHost>()) val found = Collections.synchronizedMap(linkedMapOf<String, DiscoveredHost>())
fun publish() { fun publish() {
@@ -177,9 +213,11 @@ class LanTransport(private val context: Context) : Transport {
override suspend fun join(host: DiscoveredHost, scope: CoroutineScope): Connection = override suspend fun join(host: DiscoveredHost, scope: CoroutineScope): Connection =
withContext(Dispatchers.IO) { withContext(Dispatchers.IO) {
holdRadio()
val socket = Socket() val socket = Socket()
socket.connect(InetSocketAddress(host.address, host.port), CONNECT_TIMEOUT_MS) socket.connect(InetSocketAddress(host.address, host.port), CONNECT_TIMEOUT_MS)
socket.tcpNoDelay = true socket.tcpNoDelay = true
socket.keepAlive = true
StreamConnection( StreamConnection(
peerId = host.id, peerId = host.id,
kind = TransportKind.LAN, kind = TransportKind.LAN,
@@ -207,9 +245,11 @@ class LanTransport(private val context: Context) : Transport {
stopDiscovery() stopDiscovery()
runCatching { serverSocket?.close() } runCatching { serverSocket?.close() }
serverSocket = null serverSocket = null
releaseRadio()
} }
private companion object { private companion object {
const val CONNECT_TIMEOUT_MS = 8_000 const val CONNECT_TIMEOUT_MS = 8_000
const val WIFI_LOCK_TAG = "ProLibertateGames:lan"
} }
} }
@@ -3,6 +3,7 @@ package org.prolibertate.games.net
import android.content.Context import android.content.Context
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.asStateFlow
@@ -10,6 +11,7 @@ import kotlinx.coroutines.launch
import org.prolibertate.games.game.engine.PlayerKind import org.prolibertate.games.game.engine.PlayerKind
import org.prolibertate.games.game.engine.PlayerSlot import org.prolibertate.games.game.engine.PlayerSlot
import org.prolibertate.games.game.engine.TableConfig import org.prolibertate.games.game.engine.TableConfig
import java.util.concurrent.atomic.AtomicBoolean
/** /**
* The pre-game lobby. * The pre-game lobby.
@@ -48,8 +50,27 @@ class LobbyController(
/** Peer id -> live connection, for every seated remote player. */ /** Peer id -> live connection, for every seated remote player. */
private val connections = linkedMapOf<String, Connection>() private val connections = linkedMapOf<String, Connection>()
private var clientLink: Connection? = null private var clientLink: Connection? = null
private var clientJob: Job? = null
private val jobs = mutableListOf<Job>() private val jobs = mutableListOf<Job>()
/**
* Where this device joined, so it can find its way back.
*
* The host goes on listening on the same port for as long as it is hosting,
* which is the whole match, so a guest that has lost its link can dial the
* address it already knows rather than go round discovery again.
*/
private var joinedHost: DiscoveredHost? = null
private var joinedName: String = ""
private var joinedPeerId: String = ""
private val reconnecting = AtomicBoolean(false)
/**
* The table currently being played, if any, so a replaced link can be given
* to it. Set by the match screen for as long as it is on screen.
*/
var rebinder: LinkRebinder? = null
val hostConnections: List<Connection> get() = connections.values.toList() val hostConnections: List<Connection> get() = connections.values.toList()
val clientConnection: Connection? get() = clientLink val clientConnection: Connection? get() = clientLink
@@ -118,6 +139,34 @@ class LobbyController(
private suspend fun seatPlayer(connection: Connection, hello: Hello) { private suspend fun seatPlayer(connection: Connection, hello: Hello) {
val current = _state.value val current = _state.value
val seats = current.seats.toMutableList() val seats = current.seats.toMutableList()
// A device the table already knows is not somebody new asking for a
// seat: it is a player whose link went away — a phone locked long
// enough for the radio to drop it is the usual way — coming back to the
// one it already has. Seating it again would find no free seat and turn
// it away from its own game, so it is put back where it was.
val returning = seats.indexOfFirst { it.deviceId == hello.peerId }
if (returning >= 0) {
val stale = seats[returning].peerId?.let { connections.remove(it) }
seats[returning] = seats[returning].copy(
name = hello.displayName,
kind = PlayerKind.HUMAN_REMOTE,
peerId = connection.peerId,
)
connections[connection.peerId] = connection
if (stale != null && stale !== connection) runCatching { stale.close() }
_state.value = current.copy(
seats = seats,
message = "${hello.displayName} is back.",
)
connection.send(Welcome(current.hostName, current.gameId.orEmpty(), accepted = true))
broadcastLobby()
// A match already under way keeps the seat and the position it is
// sitting in. Only the link is new.
rebinder?.rebindPeer(seats[returning].seat, connection)
return
}
val index = seats.indexOfFirst { it.kind == PlayerKind.AI } val index = seats.indexOfFirst { it.kind == PlayerKind.AI }
if (index < 0) { if (index < 0) {
connection.send( connection.send(
@@ -208,45 +257,122 @@ class LobbyController(
fun join(host: DiscoveredHost, displayName: String, peerId: String) { fun join(host: DiscoveredHost, displayName: String, peerId: String) {
jobs += scope.launch { jobs += scope.launch {
_state.value = _state.value.copy(message = "Connecting to ${host.name}…") _state.value = _state.value.copy(message = "Connecting to ${host.name}…")
val transport = transports.firstOrNull { it.kind == host.kind } ?: return@launch joinedHost = host
val connection = runCatching { transport.join(host, scope) }.getOrElse { error -> joinedName = displayName
_state.value = _state.value.copy(message = "Couldn't connect: ${error.message}") joinedPeerId = peerId
return@launch val connection = dial(host) ?: return@launch
} attachClientLink(connection)
clientLink = connection }
}
launch { private suspend fun dial(host: DiscoveredHost): Connection? {
connection.incoming.collect { message -> val transport = transports.firstOrNull { it.kind == host.kind } ?: return null
when (message) { return runCatching { transport.join(host, scope) }.getOrElse { error ->
is Welcome -> _state.value = if (message.accepted) { _state.value = _state.value.copy(message = "Couldn't connect: ${error.message}")
_state.value.copy( null
connected = true, }
gameId = message.gameId, }
hostName = message.hostName,
message = "Joined ${message.hostName}. Waiting for the host to start…",
)
} else {
_state.value.copy(message = message.reason ?: "The host declined.")
}
is LobbyUpdate -> _state.value = _state.value.copy( /**
seats = message.seats, * Starts reading [connection] as this device's way to the host, and
* introduces itself on it.
*
* Any previous reader is stopped first: collecting a connection's messages
* never ends on its own, so a link that has been replaced would otherwise
* leave its collector behind writing into the same lobby state.
*/
private suspend fun attachClientLink(connection: Connection) {
clientJob?.cancel()
clientLink = connection
clientJob = scope.launch {
connection.incoming.collect { message ->
when (message) {
is Welcome -> _state.value = if (message.accepted) {
_state.value.copy(
connected = true,
gameId = message.gameId, gameId = message.gameId,
optionsJson = message.optionsJson,
hostName = message.hostName, hostName = message.hostName,
message = "Joined ${message.hostName}. Waiting for the host to start…",
) )
} else {
is StartGame -> _state.value = _state.value.copy(started = message.config) _state.value.copy(message = message.reason ?: "The host declined.")
is Bye -> _state.value =
_state.value.copy(connected = false, message = "The host left.")
else -> Unit
} }
is LobbyUpdate -> _state.value = _state.value.copy(
seats = message.seats,
gameId = message.gameId,
optionsJson = message.optionsJson,
hostName = message.hostName,
)
is StartGame -> _state.value = _state.value.copy(started = message.config)
is Bye -> {
_state.value =
_state.value.copy(connected = false, message = "The host left.")
// A link that has gone is worth replacing while there is
// still a table to come back to. In its own job, because
// taking the new link into use stops this one.
if (rebinder != null) jobs += scope.launch { reconnect() }
}
else -> Unit
} }
} }
}
connection.send(Hello(peerId = peerId, displayName = displayName)) connection.send(Hello(peerId = joinedPeerId, displayName = joinedName))
}
/**
* Client: makes sure there is still a way to the host, and asks for the
* table again.
*
* Called when the app comes back to the front. A phone that was locked when
* its turn came round has missed whatever the host pushed in the meantime —
* a client draws nothing but what it was sent — and may have lost the link
* entirely while the radio was down, in which case there is no point asking
* politely and it dials again instead.
*
* Costs nothing when nothing is wrong: a live link is sent one [Resync],
* and the answer is the state it already has.
*/
fun restoreLink() {
if (joinedHost == null) return
jobs += scope.launch {
val live = clientLink
if (live != null && live.isOpen) live.send(Resync) else reconnect()
}
}
/**
* Dials the host again and hands the new link to whatever is being played.
*
* One attempt at a time: the link can be found to be gone twice over — the
* reader announcing it and the screen coming back both arrive at the same
* conclusion — and two dials would leave the host holding two links for one
* seat, having closed the one actually in use.
*/
private suspend fun reconnect() {
val host = joinedHost ?: return
if (!reconnecting.compareAndSet(false, true)) return
try {
repeat(RECONNECT_ATTEMPTS) { attempt ->
if (attempt > 0) delay(RECONNECT_BACKOFF_MILLIS * attempt)
if (clientLink?.isOpen == true) return
val connection = dial(host)
if (connection != null) {
// Hello first and rebind second, in that order: the host has
// to have put this device back in its seat before it can
// answer the table's resync with anything.
attachClientLink(connection)
rebinder?.rebindHostLink(connection)
return
}
}
_state.value = _state.value.copy(message = "Couldn't get back to ${host.name}.")
} finally {
reconnecting.set(false)
} }
} }
@@ -257,11 +383,28 @@ class LobbyController(
fun stop() { fun stop() {
jobs.forEach { it.cancel() } jobs.forEach { it.cancel() }
jobs.clear() jobs.clear()
clientJob?.cancel()
clientJob = null
connections.values.forEach { runCatching { it.close() } } connections.values.forEach { runCatching { it.close() } }
connections.clear() connections.clear()
runCatching { clientLink?.close() } runCatching { clientLink?.close() }
clientLink = null clientLink = null
// Nothing left to get back to, and nothing left to hand a link to.
joinedHost = null
joinedName = ""
joinedPeerId = ""
rebinder = null
transports.forEach { runCatching { it.stop() } } transports.forEach { runCatching { it.stop() } }
_state.value = State() _state.value = State()
} }
private companion object {
/**
* Enough to ride out a radio that has not finished waking up, and few
* enough that a host which is genuinely gone is reported as gone rather
* than dialled at forever.
*/
const val RECONNECT_ATTEMPTS = 4
const val RECONNECT_BACKOFF_MILLIS = 750L
}
} }
@@ -2,6 +2,7 @@ package org.prolibertate.games.net
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
@@ -56,7 +57,7 @@ class MatchController<S : Any, M : Any>(
* waits for [confirmAdvance]. * waits for [confirmAdvance].
*/ */
private val awaitsConfirmation: (S) -> Boolean = { false }, private val awaitsConfirmation: (S) -> Boolean = { false },
) { ) : LinkRebinder {
enum class Role { HOST, CLIENT } enum class Role { HOST, CLIENT }
@@ -111,15 +112,32 @@ class MatchController<S : Any, M : Any>(
private var released = false private var released = false
private val peers = mutableMapOf<String, Connection>() private val peers = mutableMapOf<String, Connection>()
/**
* The reader attached to each peer's link.
*
* Held so a replaced link's collector can be stopped. Collecting a
* [Connection.incoming] never ends on its own — a SharedFlow has no
* completion — so a link swapped out mid-match would otherwise leave its
* collector behind for the rest of the game.
*/
private val peerJobs = mutableMapOf<String, Job>()
private var hostLink: Connection? = null private var hostLink: Connection? = null
private var hostLinkJob: Job? = null
private val seatKinds: Map<Int, PlayerKind> = private val seatKinds: Map<Int, PlayerKind> =
config.seats.associate { it.seat to it.kind } config.seats.associate { it.seat to it.kind }
private val seatPeers: Map<Int, String?> =
config.seats.associate { it.seat to it.peerId } /**
* Which link each seat is reached on. Mutable because a player who drops
* and comes back is the same player in the same seat, but not necessarily
* on the same address — a reconnecting phone can be handed a different one.
*/
private val seatPeers: MutableMap<Int, String?> =
config.seats.associate { it.seat to it.peerId }.toMutableMap()
private fun nameOfPeer(peerId: String): String = private fun nameOfPeer(peerId: String): String =
config.seats.firstOrNull { it.peerId == peerId }?.name ?: "The other player" config.seats.firstOrNull { it.peerId == peerId }?.name ?: "the other player"
/** /**
* Records that this table cannot be played any further. * Records that this table cannot be played any further.
@@ -140,12 +158,7 @@ class MatchController<S : Any, M : Any>(
/** Host: begin the match and start pushing state to [clients]. */ /** Host: begin the match and start pushing state to [clients]. */
fun startAsHost(clients: List<Connection>) { fun startAsHost(clients: List<Connection>) {
check(role == Role.HOST) { "Not hosting" } check(role == Role.HOST) { "Not hosting" }
clients.forEach { connection -> clients.forEach { connection -> listenToPeer(connection) }
peers[connection.peerId] = connection
scope.launch {
connection.incoming.collect { message -> onHostMessage(connection, message) }
}
}
scope.launch { scope.launch {
val initial = runCatching { rules.initialState(config) }.getOrElse { error -> val initial = runCatching { rules.initialState(config) }.getOrElse { error ->
// A table the rules will not accept is a dead end, not a reason // A table the rules will not accept is a dead end, not a reason
@@ -162,11 +175,18 @@ class MatchController<S : Any, M : Any>(
} }
} }
private fun listenToPeer(connection: Connection) {
peers[connection.peerId] = connection
peerJobs[connection.peerId] = scope.launch {
connection.incoming.collect { message -> onHostMessage(connection, message) }
}
}
/** Client: render whatever arrives from [link]. */ /** Client: render whatever arrives from [link]. */
fun startAsClient(link: Connection) { fun startAsClient(link: Connection) {
check(role == Role.CLIENT) { "Not a client" } check(role == Role.CLIENT) { "Not a client" }
hostLink = link hostLink = link
scope.launch { hostLinkJob = scope.launch {
link.incoming link.incoming
// Asked for from inside onSubscription, which is what makes it // Asked for from inside onSubscription, which is what makes it
// reliable: the request cannot leave before this collector is // reliable: the request cannot leave before this collector is
@@ -196,11 +216,14 @@ class MatchController<S : Any, M : Any>(
is Rejected -> _notice.value = message.reason is Rejected -> _notice.value = message.reason
is Bye -> { is Bye -> {
// The host has gone: the game cannot go on, and // The link to the host has gone, either because the
// saying so is what gives the screen something to // host left or because the link died under it. The
// offer a way out of. // two are indistinguishable from this end, so this
// says only what is actually known — and it stops
// the table, which gives the screen a way out to
// offer if getting back in does not work.
_legalMoves.value = emptyList() _legalMoves.value = emptyList()
abandon("The host left the game.") abandon("The connection to the host was lost.")
} }
else -> Unit else -> Unit
@@ -209,6 +232,59 @@ class MatchController<S : Any, M : Any>(
} }
} }
// -----------------------------------------------------------------------
// Getting back in
// -----------------------------------------------------------------------
/**
* Host: a player who dropped is back, on [connection].
*
* The seat, and the position it is sitting in, were never given up — only
* the link was — so nothing here touches the game. The old link is let go
* of, the new one is read from, and the seat is sent the table as it now
* stands so the returning player has something to play on.
*/
override fun rebindPeer(seat: Int, connection: Connection) {
if (role != Role.HOST) return
val previous = seatPeers[seat]
if (previous != null) {
peerJobs.remove(previous)?.cancel()
// Not closed: the lobby owns these sockets and has already dealt
// with the one being replaced. Closing a link this end has stopped
// reading would only race the reader that is on its way out.
peers.remove(previous)
}
seatPeers[seat] = connection.peerId
listenToPeer(connection)
// Only once everybody is reachable again. At a table of four, one player
// coming back says nothing about the other one who has not.
if (seatPeers.values.filterNotNull().all { peers[it]?.isOpen == true }) {
clearAbandoned()
}
scope.launch { sendStateTo(connection) }
}
/**
* Client: the way back to the host is [connection].
*
* The collector asks for a resync as it attaches, which is what fills the
* screen back in: everything a client draws came off the wire, and whatever
* was pushed while the link was down was pushed into nothing.
*/
override fun rebindHostLink(connection: Connection) {
if (role != Role.CLIENT) return
if (hostLink === connection) return
hostLinkJob?.cancel()
clearAbandoned()
startAsClient(connection)
}
/** Puts a table that was stopped back into play. */
private fun clearAbandoned() {
_abandoned.value = null
_notice.value = null
}
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
// Input // Input
// ----------------------------------------------------------------------- // -----------------------------------------------------------------------
@@ -232,10 +308,15 @@ class MatchController<S : Any, M : Any>(
private suspend fun onHostMessage(connection: Connection, message: NetMessage) { private suspend fun onHostMessage(connection: Connection, message: NetMessage) {
if (message is Bye) { if (message is Bye) {
// One of the players has left. The host keeps the position — there is // Anything arriving on a link that has since been replaced belongs
// nothing wrong with it — but a table that cannot go on should say so // to the link, not to the seat: a player who dropped and came back
// rather than leave somebody waiting on a move that is not coming. // must not be reported as having left on the way in.
abandon("${nameOfPeer(connection.peerId)} left the game.") if (peers[connection.peerId] !== connection) return
// One of the players is no longer reachable. The host keeps the
// position — there is nothing wrong with it — but a table that
// cannot go on should say so rather than leave somebody waiting on a
// move that is not coming.
abandon("The connection to ${nameOfPeer(connection.peerId)} was lost.")
return return
} }
if (message is Resync) { if (message is Resync) {
@@ -433,8 +514,12 @@ class MatchController<S : Any, M : Any>(
* [StreamConnection.close], which this leans on to not block. * [StreamConnection.close], which this leans on to not block.
*/ */
fun close() { fun close() {
peerJobs.values.forEach { it.cancel() }
peerJobs.clear()
peers.values.forEach { runCatching { it.close() } } peers.values.forEach { runCatching { it.close() } }
peers.clear() peers.clear()
hostLinkJob?.cancel()
hostLinkJob = null
runCatching { hostLink?.close() } runCatching { hostLink?.close() }
hostLink = null hostLink = null
} }
@@ -16,7 +16,14 @@ import org.prolibertate.games.game.engine.TableConfig
* takes effect. Each client is sent a state redacted for its own seat, so an * takes effect. Each client is sent a state redacted for its own seat, so an
* opponent's hand is never transmitted in the first place. * opponent's hand is never transmitted in the first place.
*/ */
const val PROTOCOL_VERSION = 1 /**
* 2 added the heartbeat. A peer that does not answer [Ping] cannot be told
* apart from one that has gone away, so both ends have to speak it or a quiet
* link would be dropped as dead the first time nobody moved for a while. The
* lobby already refuses a mismatch with "update both devices", which is the
* honest thing to say about it.
*/
const val PROTOCOL_VERSION = 2
/** Service type and name advertised over mDNS. */ /** Service type and name advertised over mDNS. */
const val SERVICE_TYPE = "_plgames._tcp" const val SERVICE_TYPE = "_plgames._tcp"
@@ -116,6 +123,30 @@ data object Resync : NetMessage
@SerialName("rejected") @SerialName("rejected")
data class Rejected(val reason: String) : NetMessage data class Rejected(val reason: String) : NetMessage
/**
* Sent to a link nobody has spoken on for a while, and answered with [Pong].
*
* Two things need it. A phone whose screen has gone off puts its Wi-Fi radio
* into power save and, given long enough, stops the app's networking
* altogether; an access point ages an idle association out of its tables. A
* link carrying nothing between one player's turn and the next is idle for
* minutes at a time, which is exactly what both of those collect. And a TCP
* link that dies that way dies quietly: the read never returns and the next
* write is buffered into a socket that no longer goes anywhere, so without
* traffic of its own the app cannot tell a link that is waiting from one that
* is gone.
*
* Neither reaches a screen — see [StreamConnection], which answers and swallows
* them, so nothing above the transport has to know they exist.
*/
@Serializable
@SerialName("ping")
data object Ping : NetMessage
@Serializable
@SerialName("pong")
data object Pong : NetMessage
@Serializable @Serializable
@SerialName("bye") @SerialName("bye")
data object Bye : NetMessage data object Bye : NetMessage
@@ -3,11 +3,14 @@ package org.prolibertate.games.net
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.isActive import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext import kotlinx.coroutines.withContext
import kotlinx.serialization.encodeToString import kotlinx.serialization.encodeToString
import java.io.BufferedReader import java.io.BufferedReader
@@ -60,12 +63,35 @@ class StreamConnection(
private val output: OutputStream, private val output: OutputStream,
private val scope: CoroutineScope, private val scope: CoroutineScope,
private val onClosed: (StreamConnection) -> Unit = {}, private val onClosed: (StreamConnection) -> Unit = {},
/** How often a link with nothing to say is pinged. */
private val heartbeatMillis: Long = HEARTBEAT_MILLIS,
/** How long a link may go completely silent before it counts as gone. */
private val silenceMillis: Long = SILENCE_MILLIS,
) : Connection { ) : Connection {
private val reader: BufferedReader = input.bufferedReader() private val reader: BufferedReader = input.bufferedReader()
private val writer: BufferedWriter = output.bufferedWriter() private val writer: BufferedWriter = output.bufferedWriter()
private val writeLock = Mutex()
private val closed = AtomicBoolean(false) private val closed = AtomicBoolean(false)
/**
* Whether this end closed the link on purpose, as leaving a game does.
*
* Kept apart from [closed] because the two answer different questions. A
* link is closed either way; only one of the two ways is worth telling
* anybody about, and the heartbeat closes links that very much are.
*/
private val closedOnPurpose = AtomicBoolean(false)
/**
* When the peer was last heard from, on the monotonic clock.
*
* Wall time would do the wrong thing across a clock adjustment, and
* SystemClock is Android's, which the unit tests do not have.
*/
@Volatile
private var lastHeardNanos: Long = System.nanoTime()
private val _incoming = MutableSharedFlow<NetMessage>(replay = 0, extraBufferCapacity = 64) private val _incoming = MutableSharedFlow<NetMessage>(replay = 0, extraBufferCapacity = 64)
override val incoming: SharedFlow<NetMessage> = _incoming override val incoming: SharedFlow<NetMessage> = _incoming
@@ -73,8 +99,6 @@ class StreamConnection(
init { init {
scope.launch(Dispatchers.IO) { scope.launch(Dispatchers.IO) {
// Told apart from a link this side closed on purpose, because only
// one of the two is worth telling anybody about.
var droppedByPeer = false var droppedByPeer = false
try { try {
while (isActive && !closed.get()) { while (isActive && !closed.get()) {
@@ -83,11 +107,22 @@ class StreamConnection(
droppedByPeer = true droppedByPeer = true
break break
} }
// Anything at all counts as a sign of life, including a line
// this version cannot make sense of: the point is that the
// peer is still there and still talking.
lastHeardNanos = System.nanoTime()
if (line.isBlank()) continue if (line.isBlank()) continue
val message = runCatching { val message = runCatching {
protocolJson.decodeFromString<NetMessage>(line) protocolJson.decodeFromString<NetMessage>(line)
}.getOrNull() ?: continue // ignore anything we can't parse }.getOrNull() ?: continue // ignore anything we can't parse
_incoming.emit(message) // The heartbeat is the transport's own business and stops
// here. Answering on this thread is safe: send() hops to IO,
// which is where this already is.
when (message) {
is Ping -> send(Pong)
is Pong -> Unit
else -> _incoming.emit(message)
}
} }
} catch (_: Exception) { } catch (_: Exception) {
// A dropped link is normal: someone walked out of range or // A dropped link is normal: someone walked out of range or
@@ -98,30 +133,62 @@ class StreamConnection(
// lock, so it is the only one that can close it without // lock, so it is the only one that can close it without
// waiting. See close(). // waiting. See close().
runCatching { reader.close() } runCatching { reader.close() }
val weClosedIt = closed.get() val deliberate = closedOnPurpose.get()
close() shutdown()
// A link that died on its own reaches the screens on the same // A link that died on its own reaches the screens on the same
// channel as everything else, so leaving a game is something the // channel as everything else, so losing a player is something the
// other end finds out about the same way it finds out about a // other end finds out about the same way it finds out about a
// move. Withdrawing deliberately needs no announcement, and the // move. Withdrawing deliberately needs no announcement, and the
// emit is not cancellable — the whole point of it is to arrive // emit is not cancellable — the whole point of it is to arrive
// while everything around it is being torn down. // while everything around it is being torn down.
if (droppedByPeer && !weClosedIt) { if (droppedByPeer && !deliberate) {
withContext(NonCancellable) { runCatching { _incoming.emit(Bye) } } withContext(NonCancellable) { runCatching { _incoming.emit(Bye) } }
} }
} }
} }
// Keeps the link warm and, more to the point, keeps it answerable: a
// link nobody has spoken on cannot be told from a link that no longer
// exists, and between two turns of a card game that is most of the game.
scope.launch(Dispatchers.IO) {
while (isActive && !closed.get()) {
delay(heartbeatMillis)
if (closed.get()) break
val silentFor = (System.nanoTime() - lastHeardNanos) / 1_000_000
if (silentFor > silenceMillis) {
// Closing the stream is what unblocks the read; the reader
// then announces the drop from its own thread, exactly as it
// would for a peer that had hung up.
shutdown()
break
}
send(Ping)
}
}
} }
override suspend fun send(message: NetMessage) { override suspend fun send(message: NetMessage) {
if (closed.get()) return if (closed.get()) return
withContext(Dispatchers.IO) { withContext(Dispatchers.IO) {
try { // One message at a time. The framing is what makes a line a message,
writer.write(protocolJson.encodeToString<NetMessage>(message)) // and a write is three calls — the text, the newline, the flush — so
writer.newLine() // two of them running at once would splice one message through the
writer.flush() // middle of another and neither would arrive. The heartbeat is what
} catch (_: Exception) { // makes that likely: it writes on its own schedule, without caring
close() // what the game happens to be sending.
writeLock.withLock {
try {
writer.write(protocolJson.encodeToString<NetMessage>(message))
writer.newLine()
writer.flush()
} catch (_: Exception) {
// Not close(): a write that failed is a link that broke, and
// the screen has to hear about that. Going through close()
// here used to mark it deliberate, so a player whose link
// had died tapped a board that answered nothing and was
// never told why.
shutdown()
}
} }
} }
} }
@@ -143,11 +210,52 @@ class StreamConnection(
* buffered in it. * buffered in it.
*/ */
override fun close() { override fun close() {
closedOnPurpose.set(true)
shutdown()
}
/** Lets go of the streams without saying anything about why. */
private fun shutdown() {
if (!closed.compareAndSet(false, true)) return if (!closed.compareAndSet(false, true)) return
runCatching { input.close() } runCatching { input.close() }
runCatching { output.close() } runCatching { output.close() }
onClosed(this) onClosed(this)
} }
private companion object {
/**
* Often enough to hold an access point's idea of the association open
* and to notice a dead link inside a turn, seldom enough that it is a
* few dozen bytes a minute.
*/
const val HEARTBEAT_MILLIS = 5_000L
/**
* Several missed beats rather than one. A phone with its screen off
* answers late — the radio is in power save and the timer that would
* have sent the ping is deferred — and a link is not dead just because
* it is slow.
*/
const val SILENCE_MILLIS = 25_000L
}
}
/**
* A table that can be handed a replacement for a link it was given.
*
* A dropped link is not the end of a game — the position is untouched and the
* player is still sitting there — so the lobby, which owns the sockets and goes
* on listening for the whole match, reconnects and passes the new link through
* this. Narrow on purpose: the lobby has no business knowing what game is being
* played on the other side of it.
*/
interface LinkRebinder {
/** Host: [connection] is the seat's link from now on. */
fun rebindPeer(seat: Int, connection: Connection)
/** Client: [connection] is the way to the host from now on. */
fun rebindHostLink(connection: Connection)
} }
/** /**
@@ -9,6 +9,9 @@ import androidx.compose.runtime.mutableIntStateOf
import androidx.compose.runtime.remember import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue import androidx.compose.runtime.setValue
import androidx.compose.runtime.rememberCoroutineScope import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.ui.platform.LocalLifecycleOwner
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.LifecycleEventObserver
import org.prolibertate.games.game.GameCatalog import org.prolibertate.games.game.GameCatalog
import org.prolibertate.games.game.chess.ChessAi import org.prolibertate.games.game.chess.ChessAi
import org.prolibertate.games.game.chess.ChessMove import org.prolibertate.games.game.chess.ChessMove
@@ -683,6 +686,32 @@ private fun <S : Any, M : Any> StartMatch(
} }
} }
DisposableEffect(controller) { DisposableEffect(controller) {
onDispose { controller.close() } // The lobby owns the sockets and goes on listening for the whole match,
// so it is the lobby that reconnects a dropped player — and this is
// where it finds the table to hand the new link to.
env.lobby.rebinder = controller
onDispose {
env.lobby.rebinder = null
controller.close()
}
}
// Coming back to a table that has been out of sight.
//
// A locked phone is a stopped activity, and its Wi-Fi radio goes to sleep
// with the screen. Anything the host pushed in the meantime — the player's
// own turn arriving is the one that matters — was pushed at a link that may
// no longer have been there, and a client draws nothing but what it was
// sent. So the table is asked for again on the way back in, over a fresh
// link if the old one did not survive.
val lifecycleOwner = LocalLifecycleOwner.current
DisposableEffect(lifecycleOwner, route.hosting) {
val observer = LifecycleEventObserver { _, event ->
if (event == Lifecycle.Event.ON_START && !route.hosting) {
env.lobby.restoreLink()
}
}
lifecycleOwner.lifecycle.addObserver(observer)
onDispose { lifecycleOwner.lifecycle.removeObserver(observer) }
} }
} }
@@ -228,17 +228,78 @@ class MatchControllerNetworkTest {
guestLeft.start() guestLeft.start()
guestLeft.leave(Side.GUEST) guestLeft.leave(Side.GUEST)
guestLeft.settle() guestLeft.settle()
assertEquals("Guest left the game.", guestLeft.host.abandoned.value) assertEquals("The connection to Guest was lost.", guestLeft.host.abandoned.value)
val hostLeft = chess() val hostLeft = chess()
hostLeft.start() hostLeft.start()
hostLeft.leave(Side.HOST) hostLeft.leave(Side.HOST)
hostLeft.settle() hostLeft.settle()
assertEquals("The host left the game.", hostLeft.client.abandoned.value) assertEquals("The connection to the host was lost.", hostLeft.client.abandoned.value)
// Nothing more can be played on a table with no host. // Nothing more can be played on a table with no host.
assertEquals(emptyList<ChessMove>(), hostLeft.client.legalMoves.value) assertEquals(emptyList<ChessMove>(), hostLeft.client.legalMoves.value)
} }
// -----------------------------------------------------------------------
// Coming back
// -----------------------------------------------------------------------
/**
* The bug this is here for: a phone locked when its turn came round could
* not play once it was unlocked again.
*
* A locked phone puts its Wi-Fi radio to sleep, and a link carrying nothing
* between one turn and the next is exactly what that drops. The game was
* never wrong — the host still had the position and the seat was still the
* player's — but the only route to it was gone, and nothing rebuilt it. The
* board came back on unlocking with the state it had before the turn
* arrived, and every tap on it went nowhere.
*/
@Test
fun `a guest whose link died can play again once it is back`() = runBlocking {
val table = chess()
table.start()
table.white("e2", "e4")
// Twenty replies to 1. e4 — the guest is on the clock.
assertEquals(20, table.client.legalMoves.value.size)
table.dropGuestLink()
assertNotNull("A dead link has to stop the table", table.client.abandoned.value)
assertNotNull(table.host.abandoned.value)
table.reconnectGuest()
assertNull("The table is playable again and must not say otherwise", table.client.abandoned.value)
assertNull(table.host.abandoned.value)
assertNotNull("The guest came back to an empty board", table.client.state.value)
assertEquals(
"The guest is back but has nothing it is allowed to play",
20,
table.client.legalMoves.value.size,
)
// And a move made on the new link is a move that actually happens.
table.black("e7", "e5")
assertNotNull(table.host.state.value?.board?.get(squareFromName("e5")))
assertNull(table.host.state.value?.board?.get(squareFromName("e7")))
}
/**
* A guest coming back must not be treated as a guest arriving. The host end
* of the old link is the one that died; anything it has left to say about
* itself belongs to it and not to the seat, which is occupied again.
*/
@Test
fun `a guest coming back is not reported as having left`() = runBlocking {
val table = chess()
table.start()
table.dropGuestLink()
table.reconnectGuest()
table.settle()
assertNull(table.host.abandoned.value)
assertNull(table.client.abandoned.value)
}
@Test @Test
fun `closing a link on purpose is not reported as the other player leaving`() = runBlocking { fun `closing a link on purpose is not reported as the other player leaving`() = runBlocking {
val table = chess() val table = chess()
@@ -273,6 +334,11 @@ class MatchControllerNetworkTest {
private val hostScope = CoroutineScope(Job() + Dispatchers.Default) private val hostScope = CoroutineScope(Job() + Dispatchers.Default)
private val clientScope = CoroutineScope(Job() + Dispatchers.Default) private val clientScope = CoroutineScope(Job() + Dispatchers.Default)
/** Every socket a reconnection has opened since, so dispose can close them. */
private val laterSockets = mutableListOf<Socket>()
private val guestPeerId: String
private val guestSeat: Int
val hostLink: StreamConnection val hostLink: StreamConnection
private val clientLink: StreamConnection private val clientLink: StreamConnection
val host: MatchController<S, M> val host: MatchController<S, M>
@@ -281,7 +347,8 @@ class MatchControllerNetworkTest {
init { init {
clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS) clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS)
hostSocket = server.accept() hostSocket = server.accept()
val guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest" guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest"
guestSeat = (0 until seatCount).first { it != hostSeat }
hostLink = StreamConnection( hostLink = StreamConnection(
peerId = guestPeerId, peerId = guestPeerId,
@@ -332,11 +399,7 @@ class MatchControllerNetworkTest {
) )
host = controller(MatchController.Role.HOST, hostSeat, hostScope) host = controller(MatchController.Role.HOST, hostSeat, hostScope)
client = controller( client = controller(MatchController.Role.CLIENT, guestSeat, clientScope)
MatchController.Role.CLIENT,
(0 until seatCount).first { it != hostSeat },
clientScope,
)
// The lobby is already reading the guest's link when the game starts, // The lobby is already reading the guest's link when the game starts,
// and goes on reading it for the whole match. Without this the test // and goes on reading it for the whole match. Without this the test
@@ -372,6 +435,54 @@ class MatchControllerNetworkTest {
settle() settle()
} }
/**
* The guest's link dies under it, as a sleeping radio drops one.
*
* The sockets are killed rather than the connections closed: a link
* either end let go of on purpose is a player leaving, which is a
* different thing entirely and is already tested above.
*/
suspend fun dropGuestLink() {
runCatching { hostSocket.close() }
runCatching { clientSocket.close() }
settle()
}
/**
* The guest dials the host again, which is still listening on the same
* port, and both ends are handed the new link — the host from its lobby
* on the guest's Hello, the guest from its own once that has gone
* through.
*/
suspend fun reconnectGuest() {
val socket = Socket()
socket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS)
val accepted = server.accept()
laterSockets += socket
laterSockets += accepted
val newHostLink = StreamConnection(
peerId = guestPeerId,
kind = TransportKind.LAN,
input = accepted.getInputStream(),
output = accepted.getOutputStream(),
scope = hostScope,
onClosed = { runCatching { accepted.close() } },
)
val newClientLink = StreamConnection(
peerId = "lan:host",
kind = TransportKind.LAN,
input = socket.getInputStream(),
output = socket.getOutputStream(),
scope = clientScope,
onClosed = { runCatching { socket.close() } },
)
host.rebindPeer(guestSeat, newHostLink)
client.rebindHostLink(newClientLink)
settle()
}
/** Exactly what AppRoot.pop() and PlayScreen's onDispose do, in that order. */ /** Exactly what AppRoot.pop() and PlayScreen's onDispose do, in that order. */
fun leave(side: Side) { fun leave(side: Side) {
if (side == Side.HOST) { if (side == Side.HOST) {
@@ -392,6 +503,7 @@ class MatchControllerNetworkTest {
fun dispose() { fun dispose() {
runCatching { hostSocket.close() } runCatching { hostSocket.close() }
runCatching { clientSocket.close() } runCatching { clientSocket.close() }
laterSockets.forEach { runCatching { it.close() } }
runCatching { server.close() } runCatching { server.close() }
runCatching { hostLink.close() } runCatching { hostLink.close() }
runCatching { clientLink.close() } runCatching { clientLink.close() }
@@ -0,0 +1,181 @@
package org.prolibertate.games.net
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Test
import java.io.InputStream
import java.io.OutputStream
import java.net.InetSocketAddress
import java.net.ServerSocket
import java.net.Socket
import java.util.Collections
import java.util.concurrent.CountDownLatch
/**
* The heartbeat, which exists because a link with nothing on it cannot be told
* from a link that is no longer there.
*
* Between one turn of a card game and the next, nothing at all crosses the
* socket. A phone whose screen has gone off puts its Wi-Fi radio into power
* save and, given long enough, the link goes with it — and TCP does not say so:
* the read simply never returns, and the next write disappears into a socket
* that goes nowhere. So the transport talks to itself, and treats a peer that
* has stopped answering as gone.
*/
class StreamConnectionHeartbeatTest {
private val scopes = mutableListOf<CoroutineScope>()
private val closeables = mutableListOf<() -> Unit>()
@After
fun tearDown() {
closeables.forEach { runCatching { it() } }
scopes.forEach { it.cancel() }
}
private fun scope(): CoroutineScope =
CoroutineScope(Job() + Dispatchers.Default).also { scopes += it }
// -----------------------------------------------------------------------
/**
* The case the whole thing has to get right: two players thinking, nobody
* moving, for far longer than a link is allowed to be silent. Both ends
* answer each other's pings, so neither concludes the other has gone.
*/
@Test
fun `a quiet link between two live peers is left alone`() = runBlocking {
val server = ServerSocket(0)
val clientSocket = Socket()
closeables += { server.close() }
closeables += { clientSocket.close() }
clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS)
val hostSocket = server.accept()
closeables += { hostSocket.close() }
val hostScope = scope()
val clientScope = scope()
val a = connection("host", hostSocket.getInputStream(), hostSocket.getOutputStream(), hostScope)
val b = connection("guest", clientSocket.getInputStream(), clientSocket.getOutputStream(), clientScope)
val heardByA = subscribe(a, hostScope)
val heardByB = subscribe(b, clientScope)
// Several times the silence allowance, with not one game message sent.
delay(SILENCE_MILLIS * 4)
assertTrue("The host's end gave up on a peer that was answering", a.isOpen)
assertTrue("The guest's end gave up on a peer that was answering", b.isOpen)
// And the pings are the transport's own business: nothing above it sees
// them, so a screen is not asked to redraw twice a second for nothing.
assertEquals(emptyList<NetMessage>(), heardByA)
assertEquals(emptyList<NetMessage>(), heardByB)
}
/**
* A link that has died without saying so. The reader would wait on it for
* as long as the game lasted, which is what left a player tapping a board
* that answered nothing.
*/
@Test
fun `a peer that stops answering is given up on and reported`() = runBlocking {
val scope = scope()
val input = BlockingInputStream()
closeables += { input.close() }
val link = connection("gone", input, DiscardingOutputStream(), scope)
val heard = subscribe(link, scope)
// Long enough for the silence to be noticed, and then some.
delay(SILENCE_MILLIS * 3)
assertFalse("A link nobody is answering on was left open", link.isOpen)
assertEquals(
"The screen was never told the link had gone",
listOf<NetMessage>(Bye),
heard,
)
}
/** A link this end let go of is not somebody else going away. */
@Test
fun `closing a link on purpose says nothing`() = runBlocking {
val scope = scope()
val input = BlockingInputStream()
closeables += { input.close() }
val link = connection("ours", input, DiscardingOutputStream(), scope)
val heard = subscribe(link, scope)
link.close()
delay(SILENCE_MILLIS)
assertFalse(link.isOpen)
assertEquals(emptyList<NetMessage>(), heard)
}
// -----------------------------------------------------------------------
private fun connection(
peerId: String,
input: InputStream,
output: OutputStream,
scope: CoroutineScope,
) = StreamConnection(
peerId = peerId,
kind = TransportKind.LAN,
input = input,
output = output,
scope = scope,
heartbeatMillis = HEARTBEAT_MILLIS,
silenceMillis = SILENCE_MILLIS,
)
/** Everything a screen would have been shown, in order. */
private fun subscribe(link: Connection, scope: CoroutineScope): List<NetMessage> {
val heard = Collections.synchronizedList(mutableListOf<NetMessage>())
scope.launch { link.incoming.collect { heard += it } }
return heard
}
/**
* A stream that has nothing to say and never will, which is what a link
* whose far end has vanished looks like from here. Blocks until it is
* closed, then fails the read the way a closed socket does.
*/
private class BlockingInputStream : InputStream() {
private val gate = CountDownLatch(1)
override fun read(): Int {
gate.await()
throw java.io.IOException("stream closed")
}
override fun close() = gate.countDown()
}
private class DiscardingOutputStream : OutputStream() {
override fun write(b: Int) = Unit
}
private companion object {
const val CONNECT_MILLIS = 2_000
/**
* The real intervals are seconds; these are the same arrangement run
* fast, so the tests take a moment rather than a minute.
*/
const val HEARTBEAT_MILLIS = 40L
const val SILENCE_MILLIS = 200L
}
}