neiam /apollos_crib

It's Apollo's Crib

Stop a retired connection from having the last word

98e5282 · Gabriel Morell · 2026-09-02 22:03

11 files +29 -6

Message

{commit_body(@commit)}

modified clients/android/app/src/main/java/io/neiam/apolloscrib/mqtt/AnkyraClient.kt

+29 −6

@@ -29,6 +29,20 @@ class AnkyraClient(

29	29
30	30	private var client: Mqtt3AsyncClient? = null
31	31
	32	+ /**
	33	+ * Which connection attempt is the live one.
	34	+ *
	35	+ * A replaced client goes on firing its listeners: HiveMQ reconnects on its
	36	+ * own, and tearing one down produces a disconnect event that arrives after
	37	+ * its replacement has already connected. Both write the same state, so the
	38	+ * late one wins and the screen sits on "Connecting" while payloads arrive.
	39	+ * Every listener checks it still belongs to the current attempt.
	40	+ *
	41	+ * Written from the main thread, read from HiveMQ's -- hence volatile.
	42	+ */
	43	+ @Volatile
	44	+ private var generation = 0
	45	+
32	46	fun connect() {
33	47	if (!settings.isConfigured) {
34	48	onState(State.Failed)

@@ -37,6 +51,8 @@ class AnkyraClient(

37	51	disconnect()
38	52	onState(State.Connecting)
39	53
	54	+ val attempt = ++generation
	55	+
40	56	val built = MqttClient.builder()
41	57	.useMqttVersion3()
42	58	.identifier(settings.clientId)

@@ -52,12 +68,15 @@ class AnkyraClient(

52	68	// reconnect to carry the old one. A repeat subscription to the
53	69	// same filter replaces it at the broker, so this costs nothing
54	70	// and removes the case where we are connected and deaf.
55		− client?.let { subscribe(it) }
	71	+ if (attempt == generation) client?.let { subscribe(it, attempt) }
56	72	}
57	73	.addDisconnectedListener { context ->
58		− onState(
59		− if (context.reconnector.isReconnect) State.Connecting else State.Disconnected
60		− )
	74	+ if (attempt == generation) {
	75	+ onState(
	76	+ if (context.reconnector.isReconnect) State.Connecting
	77	+ else State.Disconnected
	78	+ )
	79	+ }
61	80	}
62	81	.apply { if (settings.useTls) sslWithDefaultConfig() }
63	82	.buildAsync()

@@ -80,7 +99,7 @@ class AnkyraClient(

80	99	}
81	100	.send()
82	101	.whenComplete { _, error ->
83		− if (error != null) {
	102	+ if (error != null && attempt == generation) {
84	103	// Not fatal: the reconnector keeps trying, and the
85	104	// listeners above will say so when one lands.
86	105	Log.w(TAG, "connect failed", error)

@@ -91,7 +110,7 @@ class AnkyraClient(

91	110	}
92	111	}
93	112
94		− private fun subscribe(client: Mqtt3AsyncClient) {
	113	+ private fun subscribe(client: Mqtt3AsyncClient, attempt: Int) {
95	114	client.subscribeWith()
96	115	.topicFilter(settings.topic)
97	116	// QoS 0, matching the queue name Ankyra looks for when it counts

@@ -103,6 +122,7 @@ class AnkyraClient(

103	122	}
104	123	.send()
105	124	.whenComplete { _, error ->
	125	+ if (attempt != generation) return@whenComplete
106	126	if (error != null) {
107	127	Log.w(TAG, "subscribe failed", error)
108	128	onState(State.Failed)

@@ -113,6 +133,9 @@ class AnkyraClient(

113	133	}
114	134
115	135	fun disconnect() {
	136	+ // Retires the current attempt, so whatever the old client says on its
	137	+ // way out lands after its own generation has passed and is ignored.
	138	+ generation++
116	139	client?.let { existing ->
117	140	runCatching { existing.disconnect() }
118	141	}

Parents: e1ee558