My program fetches input from a RabbitMQ or Amazaon SQS-like queue. There is not always input available in the queue. If a message is available, it triggers work. If that work is successful, message needs to be deleted so that work is not repeated. If work is unsuccessful, message needs to be made visible again, so that it is re-processed ASAP.

I'd like to use the Loaner pattern to avoid a lot of clunky code like:

checkQueue.flatMap { case None => /* no work... log and bail */ case Some(work) => try { /* a bunch of work */ } finally { deleteMessage or makeMessageVisible} }

(Doesn't look so clunky written out in psuedo-scala, but it quickly becomes clunky. Also, it's repeated all over the place.)

I'd like to use the Loaner pattern to factor out repetition and make the code more re-usable. The ultimate goal would be a a function something like this:

withQueueItem[A](block: QueueItem => \/[Rollback, Commit[A]]) : Task[A]

If no work is available, block is not executed. If work is available, block executes on it and returns Rollback or CommitA depending on whether item should be deleted from queue because it's been successfully processed.

Problems emerge right away, for example: I need a return when no work is available. I could do Task[Option[A]] or Task[/[ResourceUnavailable,A]] or any number of other things.

Another example: block has no result if it's doing a rollback... I guess I could just make Rollback Rollback(e: Throwable) and change the return type to:

 Task[\/[ResourceUnavailable, \/[Throwable, A]].

Third example: What do I want to do when deleting message from queue fails? Throw an exception? This is a case I need to catch, because if work has truly been completing it, re-doing it effects external systems. Since it needs to be caught and accounted for in the business logic, it seems like I would need to further complicate the method signature to account for...

 Task[\/[CommitFailed, \/[ResourceUnavailable, \/[Throwable, A]]]] ??????

The code below is one attempted implementation of Loaner. Each one just ends up as a tangled mess, as you can see.

//This is a horrible mess... I could use some feedback both on this specific sample, and general approach.

    abstract class WorkQueueManager[I, M] extends WithLogging

      //Check a queue (like an SQS queue) and return head message if available
      //I is body of the message deserialized to format we're actually working with
      //M is type like sqs.Message that contains info about raw item necessary for deleting it 
      protected abstract def pullNextWorkItem : Task[Option[(I, M)]]

      //With an sqs-like queue, this deletes an item form the queue
      protected abstract def markWorkItemComplete(itemMetadata: M) : Task[Unit]

      //With an sqs-like queue, this makes item visible again
      protected abstract  def forgetWorkItem(itemMetadata: M) : Task[Unit]

       //The main event
      def withWorkItem[A](block: I => Task[\/[Rollback, Commit[A]]]) : Task[\/[ResourceUnavailable, \/[Throwable, A]]] =  {

        pullNextWorkItem.flatMap {
          case None =>
            logger.debug("No work available at the moment...")
            Task.now(-\/(ResourceUnavailable("work queue item")))

          case Some((workItem: I, metadata: M)) =>
            block(workItem).map {
              case -\/(Rollback(e: Throwable)) =>
                logger.debug("Work failed 'rolling back' work item (returning to queue)")
                forgetWorkItem(metadata).runAsync {
                  case -\/(e: Throwable) =>
                  case \/-(Unit) =>

              case \/-(Commit(result: A)) =>
                logger.debug("Work succeeded. 'committing' work item (deleting from queue)")
                markWorkItemComplete(metadata).runAsync {
                  case -\/(e: Throwable) =>
                  case \/-(Unit) =>





Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.