Check if media processing finished before sending status (#2458)

* make MastodonApi.createStatus suspending

* check if media processing has finished before sending status

* add backoff for retrying processed media check
This commit is contained in:
Konrad Pozniak 2022-04-28 20:37:31 +02:00 committed by GitHub
parent 6062ec6b9e
commit 671d2c6a45
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 103 additions and 83 deletions

View file

@ -251,13 +251,15 @@ class ComposeViewModel @Inject constructor(
val sendObservable = media val sendObservable = media
.filter { items -> items.all { it.uploadPercent == -1 } } .filter { items -> items.all { it.uploadPercent == -1 } }
.map { .map {
val mediaIds = ArrayList<String>() val mediaIds: MutableList<String> = mutableListOf()
val mediaUris = ArrayList<Uri>() val mediaUris: MutableList<Uri> = mutableListOf()
val mediaDescriptions = ArrayList<String>() val mediaDescriptions: MutableList<String> = mutableListOf()
val mediaProcessed: MutableList<Boolean> = mutableListOf()
for (item in media.value!!) { for (item in media.value!!) {
mediaIds.add(item.id!!) mediaIds.add(item.id!!)
mediaUris.add(item.uri) mediaUris.add(item.uri)
mediaDescriptions.add(item.description ?: "") mediaDescriptions.add(item.description ?: "")
mediaProcessed.add(false)
} }
val tootToSend = StatusToSend( val tootToSend = StatusToSend(
@ -276,7 +278,8 @@ class ComposeViewModel @Inject constructor(
accountId = accountManager.activeAccount!!.id, accountId = accountManager.activeAccount!!.id,
draftId = draftId, draftId = draftId,
idempotencyKey = randomAlphanumericString(16), idempotencyKey = randomAlphanumericString(16),
retries = 0 retries = 0,
mediaProcessed = mediaProcessed
) )
serviceClient.sendToot(tootToSend) serviceClient.sendToot(tootToSend)

View file

@ -156,13 +156,18 @@ interface MastodonApi {
@Field("description") description: String @Field("description") description: String
): Result<Attachment> ): Result<Attachment>
@GET("api/v1/media/{mediaId}")
suspend fun getMedia(
@Path("mediaId") mediaId: String
): Response<MediaUploadResult>
@POST("api/v1/statuses") @POST("api/v1/statuses")
fun createStatus( suspend fun createStatus(
@Header("Authorization") auth: String, @Header("Authorization") auth: String,
@Header(DOMAIN_HEADER) domain: String, @Header(DOMAIN_HEADER) domain: String,
@Header("Idempotency-Key") idempotencyKey: String, @Header("Idempotency-Key") idempotencyKey: String,
@Body status: NewStatus @Body status: NewStatus
): Call<Status> ): Result<Status>
@GET("api/v1/statuses/{id}") @GET("api/v1/statuses/{id}")
fun status( fun status(

View file

@ -100,7 +100,8 @@ class SendStatusBroadcastReceiver : BroadcastReceiver() {
accountId = account.id, accountId = account.id,
draftId = -1, draftId = -1,
idempotencyKey = randomAlphanumericString(16), idempotencyKey = randomAlphanumericString(16),
retries = 0 retries = 0,
mediaProcessed = mutableListOf()
) )
) )

View file

@ -11,6 +11,7 @@ import android.content.Intent
import android.os.Build import android.os.Build
import android.os.IBinder import android.os.IBinder
import android.os.Parcelable import android.os.Parcelable
import android.util.Log
import androidx.core.app.NotificationCompat import androidx.core.app.NotificationCompat
import androidx.core.app.ServiceCompat import androidx.core.app.ServiceCompat
import androidx.core.content.ContextCompat import androidx.core.content.ContextCompat
@ -29,13 +30,12 @@ import com.keylesspalace.tusky.network.MastodonApi
import dagger.android.AndroidInjection import dagger.android.AndroidInjection
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.parcelize.Parcelize import kotlinx.parcelize.Parcelize
import retrofit2.Call import retrofit2.HttpException
import retrofit2.Callback
import retrofit2.Response
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import javax.inject.Inject import javax.inject.Inject
@ -55,7 +55,7 @@ class SendStatusService : Service(), Injectable {
private val serviceScope = CoroutineScope(Dispatchers.Main + supervisorJob) private val serviceScope = CoroutineScope(Dispatchers.Main + supervisorJob)
private val statusesToSend = ConcurrentHashMap<Int, StatusToSend>() private val statusesToSend = ConcurrentHashMap<Int, StatusToSend>()
private val sendCalls = ConcurrentHashMap<Int, Call<Status>>() private val sendJobs = ConcurrentHashMap<Int, Job>()
private val notificationManager by lazy { getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager } private val notificationManager by lazy { getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager }
@ -64,12 +64,9 @@ class SendStatusService : Service(), Injectable {
super.onCreate() super.onCreate()
} }
override fun onBind(intent: Intent): IBinder? { override fun onBind(intent: Intent): IBinder? = null
return null
}
override fun onStartCommand(intent: Intent, flags: Int, startId: Int): Int { override fun onStartCommand(intent: Intent, flags: Int, startId: Int): Int {
if (intent.hasExtra(KEY_STATUS)) { if (intent.hasExtra(KEY_STATUS)) {
val statusToSend = intent.getParcelableExtra<StatusToSend>(KEY_STATUS) val statusToSend = intent.getParcelableExtra<StatusToSend>(KEY_STATUS)
?: throw IllegalStateException("SendStatusService started without $KEY_STATUS extra") ?: throw IllegalStateException("SendStatusService started without $KEY_STATUS extra")
@ -129,82 +126,94 @@ class SendStatusService : Service(), Injectable {
statusToSend.retries++ statusToSend.retries++
val newStatus = NewStatus( sendJobs[statusId] = serviceScope.launch {
statusToSend.text, try {
statusToSend.warningText, var mediaCheckRetries = 0
statusToSend.inReplyToId, while (statusToSend.mediaProcessed.any { !it }) {
statusToSend.visibility, delay(1000L * mediaCheckRetries)
statusToSend.sensitive, statusToSend.mediaProcessed.forEachIndexed { index, processed ->
statusToSend.mediaIds, if (!processed) {
statusToSend.scheduledAt, // Mastodon returns 206 if the media was not yet processed
statusToSend.poll statusToSend.mediaProcessed[index] = mastodonApi.getMedia(statusToSend.mediaIds[index]).code() == 200
) }
}
mediaCheckRetries ++
}
} catch (e: Exception) {
Log.w(TAG, "failed getting media status", e)
retrySending(statusId)
return@launch
}
val sendCall = mastodonApi.createStatus( val newStatus = NewStatus(
"Bearer " + account.accessToken, statusToSend.text,
account.domain, statusToSend.warningText,
statusToSend.idempotencyKey, statusToSend.inReplyToId,
newStatus statusToSend.visibility,
) statusToSend.sensitive,
statusToSend.mediaIds,
statusToSend.scheduledAt,
statusToSend.poll
)
sendCalls[statusId] = sendCall mastodonApi.createStatus(
"Bearer " + account.accessToken,
account.domain,
statusToSend.idempotencyKey,
newStatus
).fold({ sentStatus ->
statusesToSend.remove(statusId)
// If the status was loaded from a draft, delete the draft and associated media files.
if (statusToSend.draftId != 0) {
draftHelper.deleteDraftAndAttachments(statusToSend.draftId)
}
val callback = object : Callback<Status> { val scheduled = !statusToSend.scheduledAt.isNullOrEmpty()
override fun onResponse(call: Call<Status>, response: Response<Status>) {
serviceScope.launch {
val scheduled = !statusToSend.scheduledAt.isNullOrEmpty() if (scheduled) {
eventHub.dispatch(StatusScheduledEvent(sentStatus))
} else {
eventHub.dispatch(StatusComposedEvent(sentStatus))
}
notificationManager.cancel(statusId)
}, { throwable ->
Log.w(TAG, "failed sending status", throwable)
if (throwable is HttpException) {
// the server refused to accept the status, save status & show error message
statusesToSend.remove(statusId) statusesToSend.remove(statusId)
saveStatusToDrafts(statusToSend)
if (response.isSuccessful) { val builder = NotificationCompat.Builder(this@SendStatusService, CHANNEL_ID)
// If the status was loaded from a draft, delete the draft and associated media files. .setSmallIcon(R.drawable.ic_notify)
if (statusToSend.draftId != 0) { .setContentTitle(getString(R.string.send_post_notification_error_title))
draftHelper.deleteDraftAndAttachments(statusToSend.draftId) .setContentText(getString(R.string.send_post_notification_saved_content))
} .setColor(
ContextCompat.getColor(
if (scheduled) { this@SendStatusService,
response.body()?.let(::StatusScheduledEvent)?.let(eventHub::dispatch) R.color.notification_color
} else {
response.body()?.let(::StatusComposedEvent)?.let(eventHub::dispatch)
}
notificationManager.cancel(statusId)
} else {
// the server refused to accept the status, save status & show error message
saveStatusToDrafts(statusToSend)
val builder = NotificationCompat.Builder(this@SendStatusService, CHANNEL_ID)
.setSmallIcon(R.drawable.ic_notify)
.setContentTitle(getString(R.string.send_post_notification_error_title))
.setContentText(getString(R.string.send_post_notification_saved_content))
.setColor(
ContextCompat.getColor(
this@SendStatusService,
R.color.notification_color
)
) )
)
notificationManager.cancel(statusId) notificationManager.cancel(statusId)
notificationManager.notify(errorNotificationId--, builder.build()) notificationManager.notify(errorNotificationId--, builder.build())
} } else {
stopSelfWhenDone() // a network problem occurred, let's retry sending the status
retrySending(statusId)
} }
} })
stopSelfWhenDone()
override fun onFailure(call: Call<Status>, t: Throwable) {
serviceScope.launch {
var backoff = TimeUnit.SECONDS.toMillis(statusToSend.retries.toLong())
if (backoff > MAX_RETRY_INTERVAL) {
backoff = MAX_RETRY_INTERVAL
}
delay(backoff)
sendStatus(statusId)
}
}
} }
}
sendCall.enqueue(callback) private suspend fun retrySending(statusId: Int) {
// when statusToSend == null, sending has been canceled
val statusToSend = statusesToSend[statusId] ?: return
val backoff = TimeUnit.SECONDS.toMillis(statusToSend.retries.toLong()).coerceAtMost(MAX_RETRY_INTERVAL)
delay(backoff)
sendStatus(statusId)
} }
private fun stopSelfWhenDone() { private fun stopSelfWhenDone() {
@ -218,8 +227,8 @@ class SendStatusService : Service(), Injectable {
private fun cancelSending(statusId: Int) = serviceScope.launch { private fun cancelSending(statusId: Int) = serviceScope.launch {
val statusToCancel = statusesToSend.remove(statusId) val statusToCancel = statusesToSend.remove(statusId)
if (statusToCancel != null) { if (statusToCancel != null) {
val sendCall = sendCalls.remove(statusId) val sendJob = sendJobs.remove(statusId)
sendCall?.cancel() sendJob?.cancel()
saveStatusToDrafts(statusToCancel) saveStatusToDrafts(statusToCancel)
@ -263,6 +272,7 @@ class SendStatusService : Service(), Injectable {
} }
companion object { companion object {
private const val TAG = "SendStatusService"
private const val KEY_STATUS = "status" private const val KEY_STATUS = "status"
private const val KEY_CANCEL = "cancel_id" private const val KEY_CANCEL = "cancel_id"
@ -319,5 +329,6 @@ data class StatusToSend(
val accountId: Long, val accountId: Long,
val draftId: Int, val draftId: Int,
val idempotencyKey: String, val idempotencyKey: String,
var retries: Int var retries: Int,
val mediaProcessed: MutableList<Boolean>
) : Parcelable ) : Parcelable