Put all Collector actions in one execute

summingbird-core/src/main/scala/com/twitter/summingbird/TimeExtractor.scala

@@ -28,4 +28,10 @@ object TimeExtractor {
     }
 }
 
-trait TimeExtractor[T] extends (T => Long) with java.io.Serializable
+/** This cannot be a subclass of function and use the pattern
+ * of implicit dependencies, since then you get an implicit function.
+ * Not good
+ */
+trait TimeExtractor[T] extends java.io.Serializable {
+  def apply(t: T): Long
+}

summingbird-online/src/main/scala/com/twitter/summingbird/online/Channel.scala

@@ -0,0 +1,147 @@
+/*
+Copyright 2013 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package com.twitter.summingbird.online
+
+import com.twitter.util.{ Await, Duration, Future, Try }
+
+import java.util.Queue
+import java.util.concurrent.{ArrayBlockingQueue, BlockingQueue, LinkedBlockingQueue, TimeUnit}
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicInteger
+/**
+ *
+ * @author Oscar Boykin
+ */
+
+object Channel {
+  /**
+   * By default, don't block on put
+   */
+  def apply[T]() = linkedNonBlocking[T]
+
+  def arrayBlocking[T](size: Int): Channel[T] =
+    fromBlocking(new ArrayBlockingQueue(size))
+
+  def linkedBlocking[T]: Channel[T] =
+    fromBlocking(new LinkedBlockingQueue())
+
+  def linkedNonBlocking[T]: Channel[T] =
+    fromQueue(new ConcurrentLinkedQueue())
+
+  def fromBlocking[T](bq: BlockingQueue[T]): Channel[T] = {
+    new Channel[T] {
+      override def add(t: T) = bq.put(t)
+      override def pollOrNull = bq.poll()
+    }
+  }
+
+  // Uses Queue.add to put. This will fail for full blocking queues
+  def fromQueue[T](q: Queue[T]): Channel[T] = {
+    new Channel[T] {
+      override def add(t: T) = q.add(t)
+      override def pollOrNull = q.poll()
+    }
+  }
+}
+
+/**
+ * Use this class with a thread-safe queue to receive
+ * results from futures in one thread.
+ * Storm needs us to touch it's code in one event path (via
+ * the execute method in bolts)
+ */
+abstract class Channel[T] {
+
+  // always increment after this
+  protected def add(t: T): Unit
+  // always decrement after this
+  protected def pollOrNull: T
+
+  private val count = new AtomicInteger(0)
+
+  def put(item: T): Int = {
+    add(item)
+    count.incrementAndGet
+  }
+
+  /** Returns the size immediately after the put */
+  def putAll(items: TraversableOnce[T]): Int = {
+    val added = items.foldLeft(0) { (cnt, item) =>
+      add(item)
+      cnt + 1
+    }
+    count.addAndGet(added)
+  }
+
+  /**
+   * check if something is ready now
+   */
+  def poll: Option[T] = pollOrNull match {
+    case null => None
+    case item =>
+        count.decrementAndGet
+        Some(item)
+  }
+
+  /**
+   * Obviously, this might not be the same by the time you
+   * call spill
+   */
+  def size: Int = count.get
+
+  // Do something on all the elements ready:
+  @annotation.tailrec
+  final def foreach(fn: T => Unit): Unit =
+    pollOrNull match {
+      case null => ()
+      case itt =>
+        count.decrementAndGet
+        fn(itt)
+        foreach(fn)
+    }
+
+  // fold on all the elements ready:
+  @annotation.tailrec
+  final def foldLeft[V](init: V)(fn: (V, T) => V): V = {
+   pollOrNull match {
+      case null => init
+      case itt =>
+        count.decrementAndGet
+        foldLeft(fn(init, itt))(fn)
+    }
+  }
+
+  /**
+   * Take enough elements to get the queue under the maxLength
+   */
+  def trimTo(maxLength: Int): Seq[T] = {
+    require(maxLength >= 0, "maxLength must be >= 0.")
+
+    @annotation.tailrec
+    def loop(size: Int, acc: List[T] = Nil): List[T] = {
+      if(size > maxLength) {
+        pollOrNull match {
+          case null => acc.reverse // someone else cleared us out
+          case item =>
+            loop(count.decrementAndGet, item::acc)
+        }
+      }
+      else acc.reverse
+    }
+    loop(count.get)
+  }
+}

summingbird-online/src/test/scala/com/twitter/summingbird/online/QueueChannelLaws.scala

@@ -0,0 +1,102 @@
+/*
+ Copyright 2013 Twitter, Inc.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+package com.twitter.summingbird.online
+
+import org.scalacheck._
+import Gen._
+import Arbitrary._
+import org.scalacheck.Prop._
+
+import com.twitter.util.{Return, Throw, Future, Try}
+
+object QueueChannelLaws extends Properties("Channel") {
+
+  property("Putting into a BoundedQueue gets size right") = forAll { (items: List[String]) =>
+    val q = Channel[String]()
+    q.putAll(items)
+    q.size == items.size
+  }
+  property("not spill if capacity is enough") = forAll { (items: List[Int]) =>
+    val q = Channel[Int]()
+    q.putAll(items)
+    q.trimTo(items.size).size == 0
+  }
+  property("Work with indepent additions") = forAll { (items: List[Int]) =>
+    val q = Channel[Int]()
+    items.map(q.put(_)) == (1 to items.size).toList
+  }
+  property("spill all with zero capacity") = forAll { (items: List[Int]) =>
+    val q = Channel[Int]()
+    q.putAll(items)
+    q.trimTo(0) == items
+  }
+  property("Channel works with finished futures") = forAll { (items: List[Int]) =>
+    val q = Channel.linkedBlocking[(Int,Try[Int])]
+    items.foreach { i => q.put((i, Try(i*i))) }
+    q.foldLeft((0, true)) { case ((cnt, good), (i, ti)) =>
+      ti match {
+        case Return(ii) => (cnt + 1, good)
+        case Throw(e) => (cnt + 1, false)
+      }
+    } == (items.size, true)
+  }
+  property("Channel.linkedNonBlocking works") = forAll { (items: List[Int]) =>
+    val q = Channel.linkedNonBlocking[(Int,Try[Int])]
+    items.foreach { i => q.put((i, Try(i*i))) }
+    q.foldLeft((0, true)) { case ((cnt, good), (i, ti)) =>
+      ti match {
+        case Return(ii) => (cnt + 1, good)
+        case Throw(e) => (cnt + 1, false)
+      }
+    } == (items.size, true)
+  }
+  property("Channel foreach works") = forAll { (items: List[Int]) =>
+    // Make sure we can fit everything
+    val q = Channel.arrayBlocking[(Int,Try[Int])](items.size + 1)
+    items.foreach { i => q.put((i,Try(i*i))) }
+    var works = true
+    q.foreach { case (i, Return(ii)) =>
+      works = works && (ii == i*i)
+    }
+    works && (q.size == 0)
+  }
+  property("Channel foldLeft works") = forAll { (items: List[Int]) =>
+    // Make sure we can fit everything
+    val q = Channel.arrayBlocking[(Int,Try[Int])](items.size + 1)
+    items.foreach { i => q.put((i,Try(i*i))) }
+    q.foldLeft(true) { case (works, (i, Return(ii))) =>
+      (ii == i*i)
+    } && (q.size == 0)
+  }
+
+  property("Channel poll + size is correct") = forAll { (items: List[Int]) =>
+    // Make sure we can fit everything
+    val q = Channel[Int]()
+    items.map { i =>
+      q.put(i)
+      val size = q.size
+      if(i % 2 == 0) {
+        // do a poll test
+        q.poll match {
+          case None => q.size == 0
+          case Some(_) => q.size == (size - 1)
+        }
+      }
+      else true
+    }.forall(identity)
+  }
+}