Move Dependants common code to graph

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

@@ -21,35 +21,7 @@ import com.twitter.summingbird.graph._
 /** Producers are Directed Acyclic Graphs
  * by the fact that they are immutable.
  */
-case class Dependants[P <: Platform[P]](tail: Producer[P, Any]) {
-  lazy val nodes: List[Producer[P, Any]] = Producer.entireGraphOf(tail)
-  lazy val allTails: List[Producer[P, Any]] = nodes.filter { fanOut(_).get == 0 }
-  private lazy val nodeSet: Set[Producer[P, Any]] = nodes.toSet
-
-  /** This is the dependants graph. Each Producer knows who it depends on
-   * but not who depends on it without doing this computation
-   */
-  private val graph: NeighborFn[Producer[P, Any]] = {
-    val nfn = Producer.dependenciesOf[P](_)
-    reversed(nodes)(nfn)
-  }
-  private val depths: Map[Producer[P, Any], Int] = {
-    val nfn = Producer.dependenciesOf[P](_)
-    dagDepth(nodes)(nfn)
-  }
-  /** The max of zero and 1 + depth of all parents if the node is the graph
-   */
-  def isNode(p: Producer[P, Any]): Boolean = nodeSet.contains(p)
-  def depth(p: Producer[P, Any]): Option[Int] = depths.get(p)
-
-  def dependantsOf(p: Producer[P, Any]): Option[List[Producer[P, Any]]] =
-    if(isNode(p)) Some(graph(p).toList) else None
-
-  def fanOut(p: Producer[P, Any]): Option[Int] = dependantsOf(p).map { _.size }
-  /**
-   * Return all dependendants of a given node.
-   * Does not include itself
-   */
-  def transitiveDependantsOf(p: Producer[P, Any]): List[Producer[P, Any]] =
-    depthFirstOf(p.asInstanceOf[Producer[P, Any]])(graph).toList
+case class Dependants[P <: Platform[P]](tail: Producer[P, Any]) extends DependantGraph[Producer[P, Any]] {
+  override lazy val nodes: List[Producer[P, Any]] = Producer.entireGraphOf(tail)
+  override def dependenciesOf(p: Producer[P, Any]) = Producer.dependenciesOf(p)
 }

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

@@ -54,11 +54,7 @@ object Producer {
   implicit def semigroup[P <: Platform[P], T]: Semigroup[Producer[P, T]] =
     Semigroup.from(_ merge _)
 
-  def dependenciesOf[P <: Platform[P]](p: Producer[P, Any]): List[Producer[P, Any]] = {
-    /*
-     * Keyed producers seem to have some issue with type inference that
-     * I work around with the cast.
-     */
+  def dependenciesOf[P <: Platform[P]](p: Producer[P, Any]): List[Producer[P, Any]] =
     p match {
       case AlsoProducer(_, prod) => List(prod)
       case NamedProducer(producer, _) => List(producer)
@@ -72,7 +68,6 @@ object Producer {
       case LeftJoinedProducer(producer, _) => List(producer)
       case Summer(producer, _, _) => List(producer)
     }
-  }
 
   /**
    * Return all dependencies of a given node in depth first, left first order.

summingbird-core/src/main/scala/com/twitter/summingbird/graph/DependantGraph.scala

@@ -0,0 +1,50 @@
+/*
+ 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.graph
+
+/** Given Dag and a List of immutable nodes, and a function to get
+ * dependencies, compute the dependants (reverse the graph)
+ */
+abstract class DependantGraph[T] {
+  def nodes: List[T]
+  def dependenciesOf(t: T): Iterable[T]
+
+  lazy val allTails: List[T] = nodes.filter { fanOut(_).get == 0 }
+  private lazy val nodeSet: Set[T] = nodes.toSet
+
+  /** This is the dependants graph. Each node knows who it depends on
+   * but not who depends on it without doing this computation
+   */
+  private lazy val graph: NeighborFn[T] = reversed(nodes)(dependenciesOf(_))
+
+  private lazy val depths: Map[T, Int] = dagDepth(nodes)(dependenciesOf(_))
+
+  /** The max of zero and 1 + depth of all parents if the node is the graph
+   */
+  def isNode(p: T): Boolean = nodeSet.contains(p)
+  def depth(p: T): Option[Int] = depths.get(p)
+
+  def dependantsOf(p: T): Option[List[T]] =
+    if(isNode(p)) Some(graph(p).toList) else None
+
+  def fanOut(p: T): Option[Int] = dependantsOf(p).map { _.size }
+  /**
+   * Return all dependendants of a given node.
+   * Does not include itself
+   */
+  def transitiveDependantsOf(p: T): List[T] = depthFirstOf(p)(graph).toList
+}