@@ -43,9 +43,11 @@ import kotlinx.coroutines.Dispatchers
4343import kotlinx.coroutines.SupervisorJob
4444import kotlinx.coroutines.delay
4545import kotlinx.coroutines.flow.first
46+ import kotlinx.coroutines.flow.mapNotNull
4647import kotlinx.coroutines.launch
4748import kotlinx.coroutines.tasks.await
4849import kotlinx.coroutines.withContext
50+ import kotlinx.coroutines.withTimeoutOrNull
4951import kotlinx.serialization.encodeToString
5052import kotlinx.serialization.json.Json
5153import okhttp3.OkHttpClient
@@ -91,13 +93,26 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
9193 }
9294 }
9395
96+ /* *
97+ * Substitutes the audio actually streamed to the watch with an already-transcoded file (see
98+ * [WatchAudioTranscoder]), bypassing [isSongTransferEligible] and the song's own local-file
99+ * resolution — the override file was just written locally by the transcoder, so it's
100+ * unconditionally eligible regardless of what the original [Song]'s source was.
101+ */
102+ data class WatchAudioOverride (
103+ val file : File ,
104+ val mimeType : String ,
105+ val bitrateBps : Int ,
106+ )
107+
94108 fun startTransferToWatch (
95109 nodeId : String ,
96110 requestId : String ,
97111 songId : String ,
98112 transferMode : String = WearTransferRequest .MODE_SAVE_TO_LIBRARY ,
99113 startPositionMs : Long = 0L,
100114 autoPlay : Boolean = false,
115+ audioOverride : WatchAudioOverride ? = null,
101116 ) {
102117 transferStateStore.markRequested(
103118 requestId = requestId,
@@ -112,6 +127,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
112127 transferMode = transferMode,
113128 startPositionMs = startPositionMs,
114129 autoPlay = autoPlay,
130+ audioOverride = audioOverride,
115131 )
116132 }
117133 }
@@ -123,6 +139,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
123139 transferMode : String ,
124140 startPositionMs : Long ,
125141 autoPlay : Boolean ,
142+ audioOverride : WatchAudioOverride ? = null,
126143 ) {
127144 var openedSongSource: OpenedSongSource ? = null
128145 try {
@@ -138,6 +155,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
138155 }
139156
140157 if (
158+ audioOverride == null &&
141159 transferMode == WearTransferRequest .MODE_SAVE_TO_LIBRARY &&
142160 ! isSongTransferEligible(song)
143161 ) {
@@ -150,19 +168,23 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
150168 return
151169 }
152170
153- val songSource = openSongSource(
154- song = song,
155- allowProxyStreaming = transferMode == WearTransferRequest .MODE_TEMPORARY_PLAYBACK ,
156- )
171+ val songSource = if (audioOverride != null ) {
172+ openOverrideSongSource(audioOverride)
173+ } else {
174+ openSongSource(
175+ song = song,
176+ allowProxyStreaming = transferMode == WearTransferRequest .MODE_TEMPORARY_PLAYBACK ,
177+ )
178+ }
157179 if (songSource == null ) {
158180 sendTransferMetadataError(
159181 nodeId = nodeId,
160182 requestId = requestId,
161183 songId = song.id,
162- errorMessage = if (transferMode == WearTransferRequest . MODE_TEMPORARY_PLAYBACK ) {
163- " Cannot stream audio source to watch "
164- } else {
165- " Cannot read audio file"
184+ errorMessage = when {
185+ audioOverride != null -> " Cannot read transcoded audio file "
186+ transferMode == WearTransferRequest . MODE_TEMPORARY_PLAYBACK -> " Cannot stream audio source to watch "
187+ else -> " Cannot read audio file"
166188 },
167189 )
168190 return
@@ -182,9 +204,9 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
182204 album = song.album,
183205 albumId = song.albumId,
184206 duration = song.duration,
185- mimeType = song.mimeType ? : " audio/mpeg" ,
207+ mimeType = audioOverride?.mimeType ? : ( song.mimeType ? : " audio/mpeg" ) ,
186208 fileSize = fileSize,
187- bitrate = song.bitrate ? : 0 ,
209+ bitrate = audioOverride?.bitrateBps ? : ( song.bitrate ? : 0 ) ,
188210 sampleRate = song.sampleRate ? : 0 ,
189211 isFavorite = song.isFavorite,
190212 paletteSeedArgb = paletteSeedArgb,
@@ -211,19 +233,12 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
211233 return
212234 }
213235
214- messageClient.sendMessage(
215- nodeId,
216- WearDataPaths .TRANSFER_METADATA ,
217- json.encodeToString(metadata).toByteArray(Charsets .UTF_8 ),
218- ).await()
219-
220- // Give the watch a brief window to reject duplicates before the audio stream starts.
221- delay(METADATA_GUARD_DELAY_MS )
222- val duplicateRejected = transferStateStore.transfers.value[requestId]?.let { transfer ->
223- transfer.status == WearTransferProgress .STATUS_FAILED &&
224- transfer.error == WearTransferProgress .ERROR_ALREADY_ON_WATCH
225- } == true
226- if (duplicateRejected) {
236+ val metadataOutcome = sendMetadataAndAwaitAck(
237+ nodeId = nodeId,
238+ requestId = requestId,
239+ metadataBytes = json.encodeToString(metadata).toByteArray(Charsets .UTF_8 ),
240+ )
241+ if (metadataOutcome == MetadataAckOutcome .DUPLICATE_REJECTED ) {
227242 runCatching { openedSongSource.close() }
228243 openedSongSource = null
229244 return
@@ -262,6 +277,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
262277 songId = song.id,
263278 inputStream = songSource.inputStream,
264279 fileSize = fileSize,
280+ transferMode = transferMode,
265281 )
266282 runCatching { songSource.close() }
267283 openedSongSource = null
@@ -325,6 +341,15 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
325341 return openHttpSongSource(streamUrl)
326342 }
327343
344+ private fun openOverrideSongSource (override : WatchAudioOverride ): OpenedSongSource ? {
345+ val file = override .file.takeIf { it.isFile && it.canRead() && it.length() > 0L } ? : return null
346+ return runCatching {
347+ OpenedSongSource (inputStream = file.inputStream(), fileSize = file.length())
348+ }.onFailure { error ->
349+ Timber .tag(TAG ).w(error, " Failed to open transcoded override file=%s" , file.absolutePath)
350+ }.getOrNull()
351+ }
352+
328353 private fun openDirectSongSource (song : Song ): OpenedSongSource ? {
329354 val directFile = song.path
330355 .takeIf { it.isNotBlank() }
@@ -739,6 +764,7 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
739764 songId : String ,
740765 inputStream : InputStream ,
741766 fileSize : Long ,
767+ transferMode : String ,
742768 ) {
743769 if (transferCancellationStore.consumeCancellation(requestId)) {
744770 sendTransferProgress(
@@ -811,14 +837,34 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
811837
812838 shouldForceCloseChannel = false
813839
814- sendTransferProgress(
815- nodeId = nodeId,
816- requestId = requestId,
817- songId = songId,
818- bytesTransferred = fileSize,
819- totalBytes = fileSize,
820- status = WearTransferProgress .STATUS_COMPLETED ,
821- )
840+ if (transferMode == WearTransferRequest .MODE_SAVE_TO_LIBRARY ) {
841+ // Bytes are on the wire, but that's not the same as "the watch has it" — wait
842+ // for the watch's own write-complete/failure report (WearCommandReceiver's
843+ // TRANSFER_PROGRESS handler, fed by WearTransferRepository) before this counts
844+ // as done. STATUS_AWAITING_WATCH_ACK is local-only, never sent to the watch — the
845+ // outer await in PlaylistWatchTransferCoordinator.transferSongToNode just keeps
846+ // waiting past it (it isn't a terminal status) until the real outcome lands, with
847+ // its own timeout as the backstop if that never happens.
848+ transferStateStore.markProgress(
849+ requestId = requestId,
850+ songId = songId,
851+ bytesTransferred = fileSize,
852+ totalBytes = fileSize,
853+ status = WearTransferProgress .STATUS_AWAITING_WATCH_ACK ,
854+ )
855+ } else {
856+ // Temporary playback isn't part of the "is this saved on watch" bookkeeping and
857+ // is latency-sensitive (the watch is waiting to start playing), so it keeps the
858+ // original optimistic-completion behavior.
859+ sendTransferProgress(
860+ nodeId = nodeId,
861+ requestId = requestId,
862+ songId = songId,
863+ bytesTransferred = fileSize,
864+ totalBytes = fileSize,
865+ status = WearTransferProgress .STATUS_COMPLETED ,
866+ )
867+ }
822868 } catch (error: Exception ) {
823869 Timber .tag(TAG ).e(error, " Failed to stream file to watch" )
824870 sendTransferProgress(
@@ -911,6 +957,58 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
911957 )
912958 }
913959
960+ private enum class MetadataAckOutcome { ACKED , DUPLICATE_REJECTED , TIMED_OUT }
961+
962+ /* *
963+ * Sends [metadataBytes] and waits for the watch to either ack receipt
964+ * ([WearTransferProgress.STATUS_METADATA_RECEIVED]) or reject it as a duplicate — instead of
965+ * the old fixed blind delay, which let the audio channel open (and the watch start draining
966+ * it) before confirming the metadata message even arrived. Real hardware testing showed that
967+ * race losing under Bluetooth contention (a connected headset, the watch's screen toggling on
968+ * a run) and the watch failing with "Transfer metadata missing" after already receiving the
969+ * whole file. Retries the send once on timeout, same shape as
970+ * [PlaylistWatchTransferCoordinator.sendPlaylistSyncToNodeWithRetry] — if the retry also times
971+ * out, streams anyway rather than blocking a whole song indefinitely; the completion-ack work
972+ * in [streamFileToWatch] still catches a genuine failure downstream.
973+ */
974+ private suspend fun sendMetadataAndAwaitAck (
975+ nodeId : String ,
976+ requestId : String ,
977+ metadataBytes : ByteArray ,
978+ ): MetadataAckOutcome {
979+ messageClient.sendMessage(nodeId, WearDataPaths .TRANSFER_METADATA , metadataBytes).await()
980+ val outcome = awaitMetadataAck(requestId)
981+ if (outcome != MetadataAckOutcome .TIMED_OUT ) return outcome
982+
983+ Timber .tag(TAG ).w(" No metadata ack from watch, retrying once: requestId=%s" , requestId)
984+ messageClient.sendMessage(nodeId, WearDataPaths .TRANSFER_METADATA , metadataBytes).await()
985+ val retryOutcome = awaitMetadataAck(requestId)
986+ if (retryOutcome == MetadataAckOutcome .TIMED_OUT ) {
987+ Timber .tag(TAG ).w(
988+ " Metadata ack still missing after retry, streaming anyway: requestId=%s" ,
989+ requestId,
990+ )
991+ }
992+ return retryOutcome
993+ }
994+
995+ private suspend fun awaitMetadataAck (requestId : String ): MetadataAckOutcome {
996+ val transfer = withTimeoutOrNull(METADATA_ACK_TIMEOUT_MS ) {
997+ transferStateStore.transfers
998+ .mapNotNull { it[requestId] }
999+ .first { transfer ->
1000+ transfer.status == WearTransferProgress .STATUS_METADATA_RECEIVED ||
1001+ (transfer.status == WearTransferProgress .STATUS_FAILED &&
1002+ transfer.error == WearTransferProgress .ERROR_ALREADY_ON_WATCH )
1003+ }
1004+ }
1005+ return when {
1006+ transfer == null -> MetadataAckOutcome .TIMED_OUT
1007+ transfer.status == WearTransferProgress .STATUS_FAILED -> MetadataAckOutcome .DUPLICATE_REJECTED
1008+ else -> MetadataAckOutcome .ACKED
1009+ }
1010+ }
1011+
9141012 private suspend fun sendTransferProgress (
9151013 nodeId : String ,
9161014 requestId : String ,
@@ -960,6 +1058,10 @@ class PhoneDirectWatchTransferCoordinator @Inject constructor(
9601058 const val TRANSFER_ARTWORK_MAX_DIMENSION = 1024
9611059 const val TRANSFER_ARTWORK_QUALITY = 95
9621060 const val TRANSFER_ARTWORK_MAX_BYTES = 1_500_000
963- const val METADATA_GUARD_DELAY_MS = 250L
1061+ // Real round-trip over the watch's Wear Data Layer connection, not a local guard delay
1062+ // (the old blind 250ms) — generous enough for Bluetooth contention with a connected
1063+ // headset, short enough that a genuinely unreachable watch fails a song quickly instead
1064+ // of dragging out a whole playlist.
1065+ const val METADATA_ACK_TIMEOUT_MS = 4_000L
9641066 }
9651067}
0 commit comments