devadmin/ProLibertateGames
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Activity

Merge pull request #31: get back into a game a locked phone dropped out of

Browse Source 
Get back into a game a locked phone dropped out of
This commit is contained in:
adrianoftyriel
2026-08-17 09:45:40 -06:00
committed by GitHub
parent d7a98c31c8 5d1d2e204e
commit 26ae3fecaa
9 changed files with 806 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/main/AndroidManifest.xml
+5
View File
@@ -6,6 +6,11 @@
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<!-- Network Service Discovery needs multicast to see hosts on the LAN. -->
<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. -->
<uses-permission android:name="android.permission.REQUEST_INSTALL_PACKAGES" />
app/src/main/java/org/prolibertate/games/net/LanTransport.kt
+40
View File
@@ -3,6 +3,7 @@ package org.prolibertate.games.net
import android.content.Context
import android.net.nsd.NsdManager
import android.net.nsd.NsdServiceInfo
import android.net.wifi.WifiManager
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.awaitClose
@@ -31,17 +32,49 @@ class LanTransport(private val context: Context) : Transport {
private val nsdManager: 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 registrationListener: NsdManager.RegistrationListener? = null
private var discoveryListener: NsdManager.DiscoveryListener? = null
private var wifiLock: WifiManager.WifiLock? = null
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
// -----------------------------------------------------------------------
override fun host(displayName: String, scope: CoroutineScope): Flow<Connection> = callbackFlow {
holdRadio()
// Port 0 lets the OS pick a free port, which is then advertised.
val server = ServerSocket(0)
serverSocket = server
@@ -66,6 +99,7 @@ class LanTransport(private val context: Context) : Transport {
while (isActive && !server.isClosed) {
val socket = runCatching { server.accept() }.getOrNull() ?: break
socket.tcpNoDelay = true
socket.keepAlive = true
val connection = StreamConnection(
peerId = socket.inetAddress?.hostAddress ?: "unknown",
kind = TransportKind.LAN,
@@ -90,6 +124,8 @@ class LanTransport(private val context: Context) : Transport {
// -----------------------------------------------------------------------
override fun discover(scope: CoroutineScope): Flow<List<DiscoveredHost>> = callbackFlow {
holdRadio()
val found = Collections.synchronizedMap(linkedMapOf<String, DiscoveredHost>())
fun publish() {
@@ -177,9 +213,11 @@ class LanTransport(private val context: Context) : Transport {
override suspend fun join(host: DiscoveredHost, scope: CoroutineScope): Connection =
withContext(Dispatchers.IO) {
holdRadio()
val socket = Socket()
socket.connect(InetSocketAddress(host.address, host.port), CONNECT_TIMEOUT_MS)
socket.tcpNoDelay = true
socket.keepAlive = true
StreamConnection(
peerId = host.id,
kind = TransportKind.LAN,
@@ -207,9 +245,11 @@ class LanTransport(private val context: Context) : Transport {
stopDiscovery()
runCatching { serverSocket?.close() }
serverSocket = null
releaseRadio()
}
private companion object {
const val CONNECT_TIMEOUT_MS = 8_000
const val WIFI_LOCK_TAG = "ProLibertateGames:lan"
}
}
app/src/main/java/org/prolibertate/games/net/LobbyController.kt
+173 -30
View File
@@ -3,6 +3,7 @@ package org.prolibertate.games.net
import android.content.Context
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
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.PlayerSlot
import org.prolibertate.games.game.engine.TableConfig
import java.util.concurrent.atomic.AtomicBoolean
/**
* The pre-game lobby.
@@ -48,8 +50,27 @@ class LobbyController(
/** Peer id -> live connection, for every seated remote player. */
private val connections = linkedMapOf<String, Connection>()
private var clientLink: Connection? = null
private var clientJob: Job? = null
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 clientConnection: Connection? get() = clientLink
@@ -118,6 +139,34 @@ class LobbyController(
private suspend fun seatPlayer(connection: Connection, hello: Hello) {
val current = _state.value
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 }
if (index < 0) {
connection.send(
@@ -208,45 +257,122 @@ class LobbyController(
fun join(host: DiscoveredHost, displayName: String, peerId: String) {
jobs += scope.launch {
_state.value = _state.value.copy(message = "Connecting to ${host.name}…")
val transport = transports.firstOrNull { it.kind == host.kind } ?: return@launch
val connection = runCatching { transport.join(host, scope) }.getOrElse { error ->
_state.value = _state.value.copy(message = "Couldn't connect: ${error.message}")
return@launch
}
clientLink = connection
joinedHost = host
joinedName = displayName
joinedPeerId = peerId
val connection = dial(host) ?: return@launch
attachClientLink(connection)
}
}
launch {
connection.incoming.collect { message ->
when (message) {
is Welcome -> _state.value = if (message.accepted) {
_state.value.copy(
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.")
}
private suspend fun dial(host: DiscoveredHost): Connection? {
val transport = transports.firstOrNull { it.kind == host.kind } ?: return null
return runCatching { transport.join(host, scope) }.getOrElse { error ->
_state.value = _state.value.copy(message = "Couldn't connect: ${error.message}")
null
}
}
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,
optionsJson = message.optionsJson,
hostName = message.hostName,
message = "Joined ${message.hostName}. Waiting for the host to start…",
)
is StartGame -> _state.value = _state.value.copy(started = message.config)
is Bye -> _state.value =
_state.value.copy(connected = false, message = "The host left.")
else -> Unit
} else {
_state.value.copy(message = message.reason ?: "The host declined.")
}
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() {
jobs.forEach { it.cancel() }
jobs.clear()
clientJob?.cancel()
clientJob = null
connections.values.forEach { runCatching { it.close() } }
connections.clear()
runCatching { clientLink?.close() }
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() } }
_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
}
}
app/src/main/java/org/prolibertate/games/net/MatchController.kt
+104 -19
View File
@@ -2,6 +2,7 @@ package org.prolibertate.games.net
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.withContext
import kotlinx.coroutines.flow.MutableStateFlow
@@ -56,7 +57,7 @@ class MatchController<S : Any, M : Any>(
* waits for [confirmAdvance].
*/
private val awaitsConfirmation: (S) -> Boolean = { false },
) {
) : LinkRebinder {
enum class Role { HOST, CLIENT }
@@ -111,15 +112,32 @@ class MatchController<S : Any, M : Any>(
private var released = false
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 hostLinkJob: Job? = null
private val seatKinds: Map<Int, PlayerKind> =
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 =
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.
@@ -140,12 +158,7 @@ class MatchController<S : Any, M : Any>(
/** Host: begin the match and start pushing state to [clients]. */
fun startAsHost(clients: List<Connection>) {
check(role == Role.HOST) { "Not hosting" }
clients.forEach { connection ->
peers[connection.peerId] = connection
scope.launch {
connection.incoming.collect { message -> onHostMessage(connection, message) }
}
}
clients.forEach { connection -> listenToPeer(connection) }
scope.launch {
val initial = runCatching { rules.initialState(config) }.getOrElse { error ->
// 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]. */
fun startAsClient(link: Connection) {
check(role == Role.CLIENT) { "Not a client" }
hostLink = link
scope.launch {
hostLinkJob = scope.launch {
link.incoming
// Asked for from inside onSubscription, which is what makes it
// 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 Bye -> {
// The host has gone: the game cannot go on, and
// saying so is what gives the screen something to
// offer a way out of.
// The link to the host has gone, either because the
// host left or because the link died under it. The
// 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()
abandon("The host left the game.")
abandon("The connection to the host was lost.")
}
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
// -----------------------------------------------------------------------
@@ -232,10 +308,15 @@ class MatchController<S : Any, M : Any>(
private suspend fun onHostMessage(connection: Connection, message: NetMessage) {
if (message is Bye) {
// One of the players has left. 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("${nameOfPeer(connection.peerId)} left the game.")
// Anything arriving on a link that has since been replaced belongs
// to the link, not to the seat: a player who dropped and came back
// must not be reported as having left on the way in.
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
}
if (message is Resync) {
@@ -433,8 +514,12 @@ class MatchController<S : Any, M : Any>(
* [StreamConnection.close], which this leans on to not block.
*/
fun close() {
peerJobs.values.forEach { it.cancel() }
peerJobs.clear()
peers.values.forEach { runCatching { it.close() } }
peers.clear()
hostLinkJob?.cancel()
hostLinkJob = null
runCatching { hostLink?.close() }
hostLink = null
}
app/src/main/java/org/prolibertate/games/net/Protocol.kt
+32 -1
View File
@@ -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
* 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. */
const val SERVICE_TYPE = "_plgames._tcp"
@@ -116,6 +123,30 @@ data object Resync : NetMessage
@SerialName("rejected")
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
@SerialName("bye")
data object Bye : NetMessage
app/src/main/java/org/prolibertate/games/net/Transport.kt
+121 -13
View File
@@ -3,11 +3,14 @@ package org.prolibertate.games.net
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import kotlinx.serialization.encodeToString
import java.io.BufferedReader
@@ -60,12 +63,35 @@ class StreamConnection(
private val output: OutputStream,
private val scope: CoroutineScope,
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 {
private val reader: BufferedReader = input.bufferedReader()
private val writer: BufferedWriter = output.bufferedWriter()
private val writeLock = Mutex()
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)
override val incoming: SharedFlow<NetMessage> = _incoming
@@ -73,8 +99,6 @@ class StreamConnection(
init {
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
try {
while (isActive && !closed.get()) {
@@ -83,11 +107,22 @@ class StreamConnection(
droppedByPeer = true
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
val message = runCatching {
protocolJson.decodeFromString<NetMessage>(line)
}.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) {
// 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
// waiting. See close().
runCatching { reader.close() }
val weClosedIt = closed.get()
close()
val deliberate = closedOnPurpose.get()
shutdown()
// 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
// move. Withdrawing deliberately needs no announcement, and the
// emit is not cancellable — the whole point of it is to arrive
// while everything around it is being torn down.
if (droppedByPeer && !weClosedIt) {
if (droppedByPeer && !deliberate) {
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) {
if (closed.get()) return
withContext(Dispatchers.IO) {
try {
writer.write(protocolJson.encodeToString<NetMessage>(message))
writer.newLine()
writer.flush()
} catch (_: Exception) {
close()
// One message at a time. The framing is what makes a line a message,
// and a write is three calls — the text, the newline, the flush — so
// two of them running at once would splice one message through the
// middle of another and neither would arrive. The heartbeat is what
// makes that likely: it writes on its own schedule, without caring
// 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.
*/
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
runCatching { input.close() }
runCatching { output.close() }
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)
}
/**
app/src/main/java/org/prolibertate/games/ui/PlayScreen.kt
+30 -1
View File
@@ -9,6 +9,9 @@ import androidx.compose.runtime.mutableIntStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue
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.chess.ChessAi
import org.prolibertate.games.game.chess.ChessMove
@@ -683,6 +686,32 @@ private fun <S : Any, M : Any> StartMatch(
}
}
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) }
}
}
app/src/test/java/org/prolibertate/games/net/MatchControllerNetworkTest.kt
+120 -8
View File
@@ -228,17 +228,78 @@ class MatchControllerNetworkTest {
guestLeft.start()
guestLeft.leave(Side.GUEST)
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()
hostLeft.start()
hostLeft.leave(Side.HOST)
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.
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
fun `closing a link on purpose is not reported as the other player leaving`() = runBlocking {
val table = chess()
@@ -273,6 +334,11 @@ class MatchControllerNetworkTest {
private val hostScope = 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
private val clientLink: StreamConnection
val host: MatchController<S, M>
@@ -281,7 +347,8 @@ class MatchControllerNetworkTest {
init {
clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS)
hostSocket = server.accept()
val guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest"
guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest"
guestSeat = (0 until seatCount).first { it != hostSeat }
hostLink = StreamConnection(
peerId = guestPeerId,
@@ -332,11 +399,7 @@ class MatchControllerNetworkTest {
)
host = controller(MatchController.Role.HOST, hostSeat, hostScope)
client = controller(
MatchController.Role.CLIENT,
(0 until seatCount).first { it != hostSeat },
clientScope,
)
client = controller(MatchController.Role.CLIENT, guestSeat, clientScope)
// 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
@@ -372,6 +435,54 @@ class MatchControllerNetworkTest {
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. */
fun leave(side: Side) {
if (side == Side.HOST) {
@@ -392,6 +503,7 @@ class MatchControllerNetworkTest {
fun dispose() {
runCatching { hostSocket.close() }
runCatching { clientSocket.close() }
laterSockets.forEach { runCatching { it.close() } }
runCatching { server.close() }
runCatching { hostLink.close() }
runCatching { clientLink.close() }
app/src/test/java/org/prolibertate/games/net/StreamConnectionHeartbeatTest.kt
+181
View File
@@ -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
}
}