Actor source

build.sbt

@@ -1,17 +1,56 @@
 name := "cornucopia"
 organization := "com.github.kliewkliew"
 
-version := "1.1.2"
+//version := "1.1.2"
+version := "0.28-SNAPSHOT"
 
 scalaVersion := "2.11.8"
 
+enablePlugins(JavaAppPackaging, DockerPlugin)
+
 resolvers += "Sonatype Releases" at "https://oss.sonatype.org/service/repositories/releases/"
 resolvers += "Typesafe Releases" at "http://repo.typesafe.com/typesafe/releases/"
 
+val akkaVersion = "2.4.17"
+
+val testDependencies = Seq(
+  "com.typesafe.akka" %%  "akka-testkit" % akkaVersion  % "test",
+  "org.scalatest"     %%  "scalatest"    % "3.0.0"      % "test",
+  "org.mockito" % "mockito-all" % "1.10.19" % Test
+)
+
 libraryDependencies ++= Seq(
   "biz.paluch.redis" % "lettuce" % "5.0.0.Beta1",
   "org.scala-lang.modules" % "scala-java8-compat_2.11" % "0.8.0",
   "com.typesafe.akka" %% "akka-stream-kafka" % "0.11-RC1",
-  "com.github.kliewkliew" %% "salad" % "0.11.01",
+  "com.typesafe.akka" %% "akka-http-core"  % "2.4.11",
+  "com.typesafe.akka" %% "akka-http-experimental"  % "2.4.11",
+  "com.typesafe.akka" %% "akka-http-spray-json-experimental"  % "2.4.11",
+//  "com.github.kliewkliew" %% "salad" % "0.11.01",
+  "com.adenda" %% "salad" % "0.11.03",
   "org.slf4j" % "slf4j-log4j12" % "1.7.22"
+) ++ testDependencies
+
+// ------------------------------------------------ //
+// ------------- Docker configuration ------------- //
+// ------------------------------------------------ //
+import NativePackagerHelper._
+
+mappings in Universal ++= directory( baseDirectory.value / "src" / "main" / "resources" )
+
+javaOptions in Universal ++= Seq(
+  "-Dconfig.file=etc/container.conf",
+  "-Dlog4j.configuration=file:/usr/local/etc/log4j.properties"
 )
+
+packageName in Docker := packageName.value
+
+version in Docker := version.value
+
+dockerBaseImage := "openjdk"
+
+dockerRepository := Some("gcr.io/adenda-server-mongodb")
+
+defaultLinuxInstallLocation in Docker := "/usr/local"
+
+daemonUser in Docker := "root"

project/plugins.sbt

@@ -0,0 +1 @@
+addSbtPlugin("com.typesafe.sbt" %% "sbt-native-packager" % "1.0.4")

src/main/resources/application.conf

@@ -43,6 +43,17 @@ cornucopia {
 
   // Time (seconds) to wait for batches to accumulate before executing a job.
   batch.period = 5
+
+  // either `kafka` or `actor`
+  source = "actor"
+
+  // microservice type: `kafka` or `http`
+  microservice.type = "http"
+
+  http {
+    host = "localhost"
+    port = "9001"
+  }
 }
 
 kafka {

src/main/scala/com/github/kliewkliew/cornucopia/kafka/Config.scala → src/main/scala/com/github/kliewkliew/cornucopia/Config.scala

@@ -1,9 +1,11 @@
-package com.github.kliewkliew.cornucopia.kafka
+package com.github.kliewkliew.cornucopia
 
 import akka.actor.ActorSystem
 import akka.kafka.scaladsl.{Producer, Consumer => ConsumerDSL}
 import akka.kafka.{ConsumerSettings, ProducerSettings, Subscriptions}
+import akka.stream.scaladsl.Source
 import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
+import com.github.kliewkliew.cornucopia.actors.CornucopiaSource
 import com.typesafe.config.ConfigFactory
 import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
 import org.slf4j.LoggerFactory
@@ -17,6 +19,7 @@ object Config {
     val gracePeriod = config.getInt("grace.period") * 1000
     val refreshTimeout = config.getInt("refresh.timeout") * 1000
     val batchPeriod = config.getInt("batch.period").seconds
+    val source = config.getString("source")
   }
 
   object Consumer {
@@ -43,8 +46,19 @@ object Config {
       .withBootstrapServers(kafkaServers)
 
     implicit val materializer = ActorMaterializer(materializerSettings)(actorSystem)
+
+    val cornucopiaActorProps = CornucopiaSource.props
+    val cornucopiaActorSource = Source.actorPublisher[CornucopiaSource.Task](cornucopiaActorProps)
+
+    val cornucopiaKafkaSource = ConsumerDSL.plainSource(sourceSettings, subscription)
+
     val cornucopiaSource = ConsumerDSL.plainSource(sourceSettings, subscription)
+
     val cornucopiaSink = Producer.plainSink(sinkSettings)
   }
 
+  object ReshardTableConfig {
+    final implicit val ExpectedTotalNumberSlots: Int = 16384
+  }
+
 }

src/main/scala/com/github/kliewkliew/cornucopia/CornucopiaException.scala

@@ -0,0 +1,13 @@
+package com.github.kliewkliew.cornucopia
+
+object CornucopiaException {
+  sealed trait CornucopiaException {
+    self: Throwable =>
+    val message: String
+    val reason: Throwable
+  }
+
+  case class CornucopiaRedisConnectionException(message: String, reason: Throwable = None.orNull)
+    extends Exception(message, reason) with CornucopiaException
+}
+

src/main/scala/com/github/kliewkliew/cornucopia/Library.scala

@@ -0,0 +1,9 @@
+package com.github.kliewkliew.cornucopia
+
+import actors.CornucopiaSource
+import akka.actor._
+
+object Library {
+  val ref: ActorRef = new graph.CornucopiaActorSource().ref
+  val source = CornucopiaSource
+}

src/main/scala/com/github/kliewkliew/cornucopia/Microservice.scala

@@ -1,7 +1,22 @@
 package com.github.kliewkliew.cornucopia
 
+import com.typesafe.config.ConfigFactory
+import org.slf4j.LoggerFactory
+import com.github.kliewkliew.cornucopia.http.Server
+
 object Microservice {
   def main(args: Array[String]): Unit = {
-    new kafka.Consumer().run
+    val config = ConfigFactory.load().getConfig("cornucopia")
+    val microserviceType = config.getString("microservice.type")
+    val logger = LoggerFactory.getLogger(this.getClass)
+
+    microserviceType match {
+      case "kafka" =>
+        logger.info("Using kafka as the microservice source")
+        new graph.CornucopiaKafkaSource().run
+      case "http" =>
+        logger.info("Using http server as the microservice source")
+        Server.start
+    }
   }
 }

src/main/scala/com/github/kliewkliew/cornucopia/actors/CornucopiaSource.scala

@@ -0,0 +1,61 @@
+package com.github.kliewkliew.cornucopia.actors
+
+import akka.actor._
+import akka.stream.actor.ActorPublisher
+
+import scala.annotation.tailrec
+
+/**
+  * Copied liberally from akka documentaion on
+  * [Stream integrations](http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-integrations.html).
+  */
+object CornucopiaSource {
+  def props: Props = Props[CornucopiaSource]
+
+  final case class Task(operation: String, redisNodeIp: String, ref: Option[ActorRef] = None)
+  case object TaskAccepted
+  case object TaskDenied
+}
+
+class CornucopiaSource extends ActorPublisher[CornucopiaSource.Task] {
+  import CornucopiaSource._
+  import akka.stream.actor.ActorPublisherMessage._
+
+  val MaxBufferSize = 100
+  var buf = Vector.empty[Task]
+
+  override def receive = {
+    case _: Task if buf.size == MaxBufferSize =>
+      sender() ! TaskDenied
+    case task: Task =>
+      if (buf.isEmpty && totalDemand > 0) {
+        val task2 = task.copy(ref = Some(sender))
+        onNext(task2)
+      }
+      else {
+        val task2 = task.copy(ref = Some(sender))
+        buf :+= task2
+        deliverBuf()
+      }
+    case Request(_) =>
+      deliverBuf()
+    case Cancel =>
+      context.stop(self)
+  }
+
+  @tailrec final def deliverBuf(): Unit = {
+    if (totalDemand > 0) {
+      if (totalDemand <= Int.MaxValue) {
+        val (use, keep) = buf.splitAt(totalDemand.toInt)
+        buf = keep
+        use foreach(task => onNext(task))
+      } else {
+        val (use, keep) = buf.splitAt(Int.MaxValue)
+        buf = keep
+        use foreach(task => onNext(task))
+        deliverBuf()
+      }
+    }
+  }
+
+}