neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

Carry the credentials into the reconnect, and deliver each board once

43141c3 · Gabriel Morell · 2026-09-03 00:02

11 files +41 -13

Message

{commit_body(@commit)}

Files changed

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

+41 −13

@@ -2,10 +2,12 @@ package io.neiam.apolloscrib.mqtt

2	2
3	3	import android.util.Log
4	4	import com.hivemq.client.mqtt.MqttClient
	5	+import com.hivemq.client.mqtt.MqttGlobalPublishFilter
5	6	import com.hivemq.client.mqtt.datatypes.MqttQos
6	7	import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient
7	8	import io.neiam.apolloscrib.data.Settings
8	9	import java.nio.charset.StandardCharsets
	10	+import java.util.concurrent.TimeUnit
9	11
10	12	/**
11	13	* A subscription to one Ankyra topic.

@@ -58,12 +60,33 @@ class AnkyraClient(

58	60	.identifier(settings.clientId)
59	61	.serverHost(settings.host)
60	62	.serverPort(settings.port)
61		− .automaticReconnectWithDefaultConfig()
	63	+ // Credentials belong on the builder, not on the connect call.
	64	+ // connectWith().simpleAuth() authenticates that one CONNECT; an
	65	+ // automatic reconnect builds its own from the client's defaults
	66	+ // and carries no credentials, so the first connect succeeds and
	67	+ // every reconnect after it is refused with BAD_USER_NAME_OR_PASSWORD.
	68	+ .apply {
	69	+ if (settings.username.isNotEmpty()) {
	70	+ simpleAuth()
	71	+ .username(settings.username)
	72	+ .password(settings.password.toByteArray(StandardCharsets.UTF_8))
	73	+ .applySimpleAuth()
	74	+ }
	75	+ }
	76	+ .automaticReconnect()
	77	+ // The default ceiling is two minutes, which is a long time to
	78	+ // stare at a stale board after a tunnel. A phone changes network
	79	+ // often enough that the wait matters more than the load a retry
	80	+ // costs.
	81	+ .initialDelay(1, TimeUnit.SECONDS)
	82	+ .maxDelay(30, TimeUnit.SECONDS)
	83	+ .applyAutomaticReconnect()
62	84	// The connection outlives the call that opened it: HiveMQ
63	85	// reconnects on its own, and a state read only from the callbacks
64	86	// below would still be saying "not connected" while payloads
65	87	// arrived. These follow the connection itself.
66	88	.addConnectedListener {
	89	+ Log.d(TAG, "connected attempt=$attempt live=$generation")
67	90	// Subscribing again on every connect rather than trusting the
68	91	// reconnect to carry the old one. A repeat subscription to the
69	92	// same filter replaces it at the broker, so this costs nothing

@@ -71,6 +94,12 @@ class AnkyraClient(

71	94	if (attempt == generation) client?.let { subscribe(it, attempt) }
72	95	}
73	96	.addDisconnectedListener { context ->
	97	+ Log.d(
	98	+ TAG,
	99	+ "disconnected attempt=$attempt live=$generation " +
	100	+ "reconnect=${context.reconnector.isReconnect} " +
	101	+ "cause=${context.cause}"
	102	+ )
74	103	if (attempt == generation) {
75	104	onState(
76	105	if (context.reconnector.isReconnect) State.Connecting

@@ -82,6 +111,16 @@ class AnkyraClient(

82	111	.buildAsync()
83	112	client = built
84	113
	114	+ // Registered once, not per subscription. A subscribe callback stays
	115	+ // registered for the life of the client, so re-subscribing on every
	116	+ // reconnect stacked another copy and each board arrived once per
	117	+ // reconnect the phone had ever made.
	118	+ built.publishes(MqttGlobalPublishFilter.ALL) { publish ->
	119	+ if (attempt == generation) {
	120	+ onMessage(String(publish.payloadAsBytes, StandardCharsets.UTF_8))
	121	+ }
	122	+ }
	123	+
85	124	built.connectWith()
86	125	// The board is a snapshot, not a log: on reconnect we want what is
87	126	// true now, and the retained message plus the next tick give that.

@@ -89,14 +128,6 @@ class AnkyraClient(

89	128	// Long enough that a dozing phone is not churning the connection,
90	129	// short enough that the broker notices a dead one.
91	130	.keepAlive(60)
92		− .apply {
93		− if (settings.username.isNotEmpty()) {
94		− simpleAuth()
95		− .username(settings.username)
96		− .password(settings.password.toByteArray(StandardCharsets.UTF_8))
97		− .applySimpleAuth()
98		− }
99		− }
100	131	.send()
101	132	.whenComplete { _, error ->
102	133	if (error != null && attempt == generation) {

@@ -116,10 +147,6 @@ class AnkyraClient(

116	147	// QoS 0, matching the queue name Ankyra looks for when it counts
117	148	// who is listening -- `mqtt-subscription-<client_id>qos0`.
118	149	.qos(MqttQos.AT_MOST_ONCE)
119		− .callback { publish ->
120		− val payload = String(publish.payloadAsBytes, StandardCharsets.UTF_8)
121		− onMessage(payload)
122		− }
123	150	.send()
124	151	.whenComplete { _, error ->
125	152	if (attempt != generation) return@whenComplete

@@ -127,6 +154,7 @@ class AnkyraClient(

127	154	Log.w(TAG, "subscribe failed", error)
128	155	onState(State.Failed)
129	156	} else {
	157	+ Log.d(TAG, "subscribed to ${settings.topic}")
130	158	onState(State.Connected)
131	159	}
132	160	}

Parents: a155b6a