Thread-safe summing buffer using only ArrayBlockingQueue and Monoids.

summingbuffer.scala

// Adds an item to the buffer and returns None, or fills the buffer and returns the sum
def addOrSum(item: V): Option[V] = {
  @tailrec
  def offerAll(items: List[V], acc: Option[V] = None): Option[V] =
    items match {
      case Nil => acc
      case head :: tail => {
        if(!queue.offer(head)) {
          // Queue is full
          val toSum = ListBuffer[V]()
          queue.drainTo(toSum)
          // Add everything up and get the new acc:
          val newAcc = Monoid.plus(acc, Some(Monoid.sum(items)))
          // Now continue to offer:
          offerAll(tail, newAcc)
        }
        else {
          // We added and we should just recursively call:
          offerAll(tail, acc)
        }
      }
    }
  // Start offering everything:
  offerAll(List(item))
}