gistfile1.scala

package au.com.cba.omnia.eventually.ops

import au.com.cba.omnia.eventually.schema.{Entity, EntityAware}
import SelectEntitySyntax._

import com.twitter.scalding.typed.{ TypedPipe, Grouped }

import scala.reflect.ClassTag

trait GroupByEntity[C] {
  type K
  def op[E: EntityAware](p: TypedPipe[E]): Grouped[K,E]
}

trait LowPriorityGroupByEntity {
  implicit val groupByEntityBare = {
    new GroupByEntity[Nothing] {
      type K = Entity
      def op[E: EntityAware](p: TypedPipe[E]) = p.selectEntity[Entity].group
    }
  }
}

object GroupByEntity extends LowPriorityGroupByEntity {

  @inline def apply[A](implicit groupByEntity: GroupByEntity[A]) = groupByEntity


  implicit def groupBySelectedEntity[C <: Entity : ClassTag] = {
    new GroupByEntity[C] {
      type K = C
      def op[E: EntityAware](p: TypedPipe[E]) = p.selectEntity[C].group
    }
  }


}

trait GroupByEntitySyntax {

  implicit class GroupByEntityOps[E: EntityAware](pipe: TypedPipe[E]) {
    def groupByEntity[C: GroupByEntity] = {
      GroupByEntity[C].op[E](pipe)
    }
  }
}

object GroupByEntitySyntax extends GroupByEntitySyntax