Long polling for a sqs queue?

[u6f6o]

Hi,

I am currently implementing an email service that receives business events from an aws sqs queue, forwards them to the vertx event bus and finally calls the mailgun api to trigger the email sending. As far as I am aware of, there is currently no long polling mechanism available for sqs queues, thus I came up with my own and would like to ask, if anybody could have a look on it and share his/her opinion? I am fairly new to this topic (mutiny) and any feedback is highly appreciated.

Below you'll find my current solution:

import emailer.events.EventType.DOI_CONFIRMATION
import emailer.events.EventType.PASSWORD_REST
import io.quarkus.logging.Log
import io.quarkus.runtime.ShutdownEvent
import io.quarkus.runtime.StartupEvent
import io.smallrye.mutiny.Multi
import io.smallrye.mutiny.Uni
import io.smallrye.mutiny.coroutines.asUni
import io.smallrye.mutiny.groups.MultiRepetition
import io.smallrye.mutiny.groups.UniOnFailure
import io.smallrye.mutiny.groups.UniRepeat
import io.vertx.core.eventbus.DeliveryOptions
import io.vertx.mutiny.core.eventbus.EventBus
import io.vertx.mutiny.core.eventbus.Message
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import org.eclipse.microprofile.config.inject.ConfigProperty
import java.time.Duration
import java.util.concurrent.atomic.AtomicBoolean
import java.util.function.Supplier
import javax.enterprise.context.ApplicationScoped
import javax.enterprise.event.Observes

@ApplicationScoped
class EventPollingService(
    val sqsService: SqsService,
    @ConfigProperty(name = "application.event-polling.delay-on-failure")
    val delayDurationOnFailure: Duration,
    val eventBus: EventBus
) {

    val running = AtomicBoolean(true)

    val addressByEventType = mapOf (
        PASSWORD_REST to "password-reset",
        DOI_CONFIRMATION to "double-opt-in"
    )

    /**
     * Starts an infinite long polling mechanism that will repeatedly fetch events
     * from the emailer sqs queue. The [SqsService.receiveEvents] method is supposed
     * to delay the retrieval of empty result sets (no event currently present) by
     * a fixed amount of time (20 secs).
     *
     * In case the event retrieval results in a failure, the further retrieval of
     * events is paused by 10 seconds. If we don't to this, a failure that occurs
     * instantaneously and repeatedly might cause the application to consume all
     * available cpu resources.
     *
     * It's also important to replace a failure by a meaningful fallback because
     * otherwise the [io.smallrye.mutiny.groups.MultiCreateBy.repeating] will
     * terminate.
     *
     * In case an event processor causes a failure, we'll simply stall the event
     * and hope that it is successfully processed the next time - otherwise
     * it will end up in the dlq.
     *
     */
    fun startUp(@Observes event: StartupEvent) {
        Multi
            .createBy()
            .repeating()
            .uniSupplier {
                asUni { sqsService.receiveEvents() }
                    .onFailure()
                    .pauseFurtherProcessingFor(delayDurationOnFailure)
            }
            .whilst { running.get() }
            .subscribe()
            .with(
                { events ->
                    processEvents(events)
                },
                { failure ->
                    Log.error("Terminal error occurred, long polling will most likely be stopped.", failure)
                }
            )
    }

    fun shutDown(@Observes event: ShutdownEvent) {
        running.set(false)
    }

    private fun processEvents(events: List<SqsEvent<*>>) {
        events
            .map { event -> processEvent(event) }
            .takeIf { it.isNotEmpty() }
            ?.let {
                Uni
                    .join()
                    .all(it)
                    .andCollectFailures()
                    .subscribe()
                    .with(
                        { Log.info("Event processing succeeded.") },
                        { failure -> Log.warn("Event processing failed.", failure) }
                    )
            }
    }

    private fun processEvent(event: SqsEvent<*>): Uni<Unit> = forwardEventToEventProcessor(event)
        .onItemOrFailure()
        .transformToUni { message, throwable ->
            if (throwable == null) {
                asUni {
                    sqsService.deleteMessage(message.body())
                }
            } else {
                Log.warn(
                    "Event processing failed for $event.",
                    throwable
                )
                Uni.createFrom().nullItem() // this should lead to a retry sqs-wise
            }
        }

    private fun forwardEventToEventProcessor(event: SqsEvent<*>): Uni<Message<String>> {
        val eventType = event.metaData.eventType
        val address = addressByEventType[eventType]

        return eventBus.request(
            address,
            event,
            DeliveryOptions().setLocalOnly(true)
        )
    }

    @Suppress("UNCHECKED_CAST")
    private fun <T> MultiRepetition.uniSupplier(supplier: Supplier<Uni<T>>): UniRepeat<T> {
        return this.uni(supplier as Supplier<Uni<out T>>)
    }

    @Suppress("UNCHECKED_CAST")
    private fun <T> UniOnFailure<T>.recoverWithUniSupplier(supplier: Supplier<Uni<T>>): Uni<T> {
        return this.recoverWithUni(supplier as Supplier<Uni<out T>>)
    }

    /**
     * Pauses the further processing of an erroneous uni. This is necessary if the
     * operation the uni relies on, naturally delays the further processing
     * (e.g. waiting until events are available in a sqs queue).
     *
     * Further reading:
     * https://smallrye.io/smallrye-mutiny/1.6.0/guides/delaying-events/#delaying-other-types-of-events
     */
    private fun <T> UniOnFailure<T>.pauseFurtherProcessingFor(duration: Duration): Uni<T> {
        return this
            .transform { throwable ->
                throwable.also {
                    Log.error("Pause further event polling for $duration.", it)
                }
            }
            .onFailure()
            .recoverWithUniSupplier {
                Uni
                    .createFrom()
                    .nullItem<T>()
                    .onItem()
                    .delayIt()
                    .by(duration)
            }
    }

    private fun <T> asUni(block: suspend () -> T): Uni<T> = GlobalScope.async { block() }.asUni()
}

[bboyz269]

If I understand your use case correctly, an UniRetry would be more clean.

Uni.createFrom()
    .completionStage { sqsAsyncClient.receiveMessage(...) } // make use of AWS async clients
    .flatMap {
        // processing
        // ...
        // Always resulting in a specific failure to trigger retry
        Uni.createFrom().failure(SqSPollingInProcessException()) 
    }
    .onFailure(SqSPollingInProcessException::class.java)
    .retry()
    .withBackOff(Duration.ofSeconds(20))
    .indefinitely()