atamborrino

case class PerformanceStats(
  ioId: IOId,
  dsp: Dsp,
  advertiserId: Option[Int],
  campaignId: Int,
  campaignName: String,
  campaignCurrency: Option[String],
  strategyId: Int,
  strategyName: String,
  day: LocalDate,

atamborrino

// An ADT+shapeless as a drop-in replacement for a standard Scala Enumeration.
//
// First the unsafe standard Scala Enumeration ...
//
object ScalaEnumDemo extends App {
  // Example from scala.Enumeration scaladoc. Terse ...
  object WeekDay extends Enumeration {
    type WeekDay = Value
    val Mon, Tue, Wed, Thu, Fri, Sat, Sun = Value
  }

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")

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

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

  /**
   * 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

 /**
     * 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

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

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

  // 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).