atamborrino

// Enum type based on Sum types (OR)
trait ADTEnum[A] {
  import play.api.data.mapping.json.Writes._
  import play.api.data.mapping.json.Rules._

  val list: Seq[A]

  def withName(name: String): Option[A] = {
    list.find(_.toString.toLowerCase == name.toLowerCase)
  }

atamborrino

  // We define a logical rule Augmentable that transform Int -> Long and Float -> Double 
  // at the type level with a dependent type.
  // Then we want to abstract over arity and ensure that an "augment" transformation 
  // is defined for every type of a HList.
 
  import shapeless._

  // Dependent type T -> S (S depends on T)
  trait Augmentable[T] {
    type S

atamborrino

  // We want to use the Scala type system as a logic programming language by resolving at compile
  // the simple "ancestor" Prolog exercise.

  import shapeless._
  import syntax.singleton._

  // Prolog version

//  parent(stacy, bob).
//  parent(bob, bill).

atamborrino

object FlowHelper {
  def mapAsyncUnorderedWithBoundedParallelism[A, B](parallelism: Int)(f: A => Future[B]): Flow[A, B] =
    Flow[A].section(OperationAttributes.inputBuffer(initial = parallelism, max = parallelism)) { sectionFlow =>
      sectionFlow.mapAsyncUnordered(f)
    }

  def mapAsyncWithBoundedParallelism[A, B](parallelism: Int)(f: A => Future[B]): Flow[A, B] =
    Flow[A].section(OperationAttributes.inputBuffer(initial = parallelism, max = parallelism)) { sectionFlow =>
      sectionFlow.mapAsync(f)
    }

atamborrino

CREATE OR REPLACE FUNCTION sum_hll(a bytea, b bytea) RETURNS bytea as $$
DECLARE
	local_result bytea = a;
BEGIN
	IF get_byte(a, 0) <> get_byte(b, 0) OR get_byte(a, 1) <> get_byte(b, 1) THEN
		RAISE EXCEPTION 'HLL ERROR: FIRST 2 BYTES OF HLLs ARE NOT EQUAL. CANNOT SUM.';
	END IF;

	IF length(a) <> length(b) THEN 
		RAISE EXCEPTION 'HLL ERROR: HLLs LENGTH ARE NOT EQUAL. CANNOT SUM.';

atamborrino

 /**
     * Streamed upload of an akka stream to S3
     * @param  bucket the bucket name
     * @param  key the key of file
     * @param  source a source of array of bytes
     * @return a successful future of the uploaded number of chunks (or a failure)
     */
    def uploadStream(bucket: String, key: String, source: Source[Array[Byte]], parallelism: Int = 1)(implicit fm: FlowMaterializer): Future[Int] = {
      import scala.collection.JavaConversions._
      import client.executionContext

atamborrino

  /**
   * Conflate and/or expand the stream with an user-defined function
   * @param zero initial state
   * @param f take current state and current elem, returns a seq of C elements to push downstream and the next state b
   *          if we want the stream to continue (if no new state b, the stream ends).
   * @param lastPushIfUpstreamEnds if the upstream ends (before customStatefulProcessor decides to end the stream),
   *                               this function is called on the last b state and the resulting c elements
   *                               are pushed downstream as the last elements of the stream.
   * @return
   */

atamborrino

package object worker {

  case class QueueUrl(value: String) extends AnyVal

  trait SerDe[A] {
    def serialize(a: A): String
    def deserialize(s: String): A
  }

  trait Ackable {

atamborrino

package object worker {

  trait SerDe[MsgType] {
    def serialize(msgType: MsgType): String
    def deserialize(s: String): MsgType
  }
  
  case class QueueUrl(value: String) extends AnyVal
  
  abstract class Queue[-A](implicit serde: SerDe[A]) {

atamborrino

 def trying[O](f: String => O)(msg: String) =
    Rule.fromMapping[String, O] { s =>
      Try(f(s))
        .map(Success[ValidationError, O](_))
        .getOrElse(Failure[ValidationError, O](Seq(ValidationError(msg, s))))
    }
    
val stringToDoubleR: Rule[String, Double] = trying(_.toDouble)("error.double")
val stringToLongR: Rule[String, Long] = trying(_.toLong)("error.long")
val stringToIntR: Rule[String, Int] = trying(_.toInt)("error.int")