Work on clustering v2

README.md

@@ -11,20 +11,20 @@ The easiest way to run the scraper with stand-alone mode is to use the neat CLI
 sbt assembly
 
 # Run it with 
-java -jar target/*/scraper.jar --categories prodaja --pages 10
+java -jar scraper/*/*/scraper.jar --categories prodaja --pages 10
 ```
 
 By default, the scraper spits out [JSON]. 
 
 ```bash
-java -jar target/*/scraper.jar --categories prodaja --pages 2 | jq -R 'fromjson?'
+java -jar scraper/*/*/scraper.jar --categories prodaja --pages 2 | jq -R 'fromjson?'
 ```
 
 So to make things bit easier for your eyes your 
 can use [jq] to format or restructure output further for example to CSV.
 
 ```bash
-java -jar target/*/scraper.jar --categories prodaja --pages 10 \
+java -jar scraper/*/*/scraper.jar --categories prodaja --pages 10 \
     | jq -R 'fromjson?' \
     | jq -r "([.refNumber, .title, .price, .location.latitude, .location.longitude]) | @csv" \
     > prodaja.csv
@@ -33,7 +33,7 @@ java -jar target/*/scraper.jar --categories prodaja --pages 10 \
 Adjusting parallelism and other [fine `application.conf` switches][configuration] can be easily done via loading of different configuration.
 
 ```bash
-java -Dconfig.resource=quick.conf -jar target/*/scraper.jar --categories najem
+java -Dconfig.resource=quick.conf -jar scraper/*/*/scraper.jar --categories najem
 ```
 
 Some [configuration options][configuration] can also be adjusted via environment variables i.e.
@@ -43,10 +43,15 @@ INITIAL_CATEGORIES=prodaja,najem
 CATEGORY_PAGES_LIMIT=3
 ```
 
+## Resources
+
+- https://blog.softwaremill.com/running-akka-cluster-on-kubernetes-e4cd2913e951
+
+
 ## Authors
 
 - [Oto Brglez](https://github.com/otobrglez)
 
-[configuration]: src/main/resources/application.conf
+[configuration]: scraper/src/main/resources/application.conf
 [jq]: https://stedolan.github.io/jq/
 [JSON]: https://www.json.org/json-en.html

build.sbt

@@ -1,60 +1,33 @@
-name := "pinkstack-realestate"
-
-version := "0.0.1"
-
-scalaVersion := "2.13.2"
-
-libraryDependencies ++= Seq(
-  // Akka and Akka Streams
-  "com.typesafe.akka" %% "akka-stream" % "2.6.6",
-  "com.typesafe.akka" %% "akka-http" % "10.1.12",
-
-  // FP
-  "org.typelevel" %% "cats-core" % "2.0.0",
-
-  // Lenses
-  "com.github.julien-truffaut" %% "monocle-core" % "2.0.3",
-  "com.github.julien-truffaut" %% "monocle-macro" % "2.0.3",
-  "com.github.julien-truffaut" %% "monocle-law" % "2.0.3" % "test",
-
-  // Configuration
-  "com.typesafe" % "config" % "1.4.0",
-  "com.github.pureconfig" %% "pureconfig" % "0.12.3",
-
-  // URL Parsing
-  "io.lemonlabs" %% "scala-uri" % "2.2.2",
-
-  // CLI
-  "com.monovore" %% "decline" % "1.2.0",
-
-  // Parsing
-  "org.jsoup" % "jsoup" % "1.13.1",
-
-  // JSON
-  "io.circe" %% "circe-core" % "0.12.3",
-  "io.circe" %% "circe-generic" % "0.12.3",
-  "io.circe" %% "circe-parser" % "0.12.3",
-
-  // Logging
-  "ch.qos.logback" % "logback-classic" % "1.2.3",
-  "com.typesafe.scala-logging" %% "scala-logging" % "3.9.2",
-  "com.typesafe.akka" %% "akka-slf4j" % "2.6.6",
-
-  // Testing
-  "org.scalatest" %% "scalatest" % "3.2.0" % "test",
-  "org.scalatest" %% "scalatest-flatspec" % "3.2.0" % "test",
-  "org.scalatest" %% "scalatest-shouldmatchers" % "3.2.0" % "test"
-)
-
-scalacOptions ++= Seq(
-  "-deprecation",
-  "-encoding", "UTF-8",
-  "-feature",
-  "-target:jvm-1.11",
-  "-unchecked"
-)
-
-enablePlugins(JavaAppPackaging)
-
-mainClass in assembly := Some("com.pinkstack.realestate.apps.Scraper")
-assemblyJarName in assembly := "scraper.jar"
+import Dependencies._
+import Settings._
+import scala.sys.process._
+
+lazy val scraper = (project in file("scraper"))
+  .settings(sharedSettings: _*)
+  .settings(
+    name := "scraper",
+    Compile / mainClass := Some(""),
+    libraryDependencies ++=
+      akka ++ akkaHttp ++ cats
+        ++ monocle ++ configurationLibs ++ scalaUri
+        ++ decline ++ jsoup ++ circe
+        ++ akkaHttpCirce ++ logging ++ scalaTest,
+    Compile / mainClass := Some("com.pinkstack.realestate.scraper.apps.Scraper"),
+  )
+  .enablePlugins(JavaAppPackaging)
+  .settings(assemblyJarName in assembly := "scraper.jar")
+
+lazy val cluster = (project in file("cluster"))
+  .settings(sharedSettings: _*)
+  .settings(
+    name := "cluster",
+    Compile / mainClass := Some(""),
+    libraryDependencies ++=
+      akka ++ akkaHttp ++ cats
+        ++ monocle ++ configurationLibs ++ scalaUri
+        ++ decline ++ jsoup ++ circe
+        ++ akkaHttpCirce ++ logging ++ scalaTest
+  )
+  .enablePlugins(JavaAppPackaging)
+  .settings(assemblyJarName in assembly := "cluster.jar")
+  .dependsOn(scraper)

cluster/src/main/scala/com/pinkstack/realestate/cluster/NodeApp.scala

@@ -0,0 +1,53 @@
+/**
+ * Author: Oto Brglez - <otobrglez@gmail.com>
+ */
+package com.pinkstack.realestate.cluster
+
+import akka.actor.typed._
+import akka.actor.typed.scaladsl._
+import akka.cluster.ClusterEvent.MemberEvent
+import akka.cluster.typed._
+import com.monovore.decline._
+import com.typesafe.config.{Config, ConfigFactory}
+
+object ClusterListener {
+
+  sealed trait Event
+
+  case class ClusterChange(event: MemberEvent) extends Event
+
+  def apply(): Behavior[Event] = Behaviors.setup[Event] { context =>
+    val clusterEvents: ActorRef[MemberEvent] = context.messageAdapter(ClusterChange)
+    Cluster(context.system).subscriptions ! Subscribe(clusterEvents, classOf[MemberEvent])
+
+    Behaviors.receiveMessage {
+      case ClusterChange(event: MemberEvent) =>
+        context.log.info("🐝 {}", event)
+        Behaviors.same
+    }
+  }
+}
+
+object Node {
+  def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] { context =>
+    context.spawn(ClusterListener(), "ClusterListener")
+    Behaviors.empty
+  }
+}
+
+object System {
+  val configuration: Int => Config = port =>
+    ConfigFactory.parseString(
+      s"""akka.remote.artery.canonical.port=$port""".stripMargin)
+      .withFallback(ConfigFactory.load("clusteringV4.conf"))
+
+  def apply(port: Int): ActorSystem[Nothing] =
+    ActorSystem[Nothing](Node(), "AppV4", configuration(port))
+}
+
+object NodeApp extends CommandApp(
+  name = "node", header = "Boots up a single (seed) cluster node",
+  main = for {
+    port <- Opts.option[Int]("port", "port number").withDefault(0)
+  } yield System(port)
+)

project/Dependencies.scala

@@ -0,0 +1,98 @@
+import sbt._
+
+object Dependencies {
+
+  object V {
+    val akka = "2.6.6"
+    val akkaHttp = "10.1.12"
+
+    val cats = "2.0.0"
+
+    val monocle = "2.0.3"
+
+    val typesafeConfig = "1.4.0"
+    val pureConfig = "0.12.3"
+
+    val scalaUri = "2.2.2"
+
+    val decline = "1.2.0"
+
+    val jsoup = "1.13.1"
+    val circe = "0.12.3"
+
+    val akkaHttpCirce = "1.31.0"
+    val logbackClassic = "1.2.3"
+    val scalaLogging = "3.9.2"
+    val akkaSlf4j = "2.6.6"
+
+    val scalaTest = "3.2.0"
+  }
+
+  lazy val akka: Seq[ModuleID] = Seq(
+    "com.typesafe.akka" %% "akka-actor-typed" % V.akka,
+    "com.typesafe.akka" %% "akka-stream" % V.akka,
+
+    "com.typesafe.akka" %% "akka-cluster" % V.akka,
+    "com.typesafe.akka" %% "akka-cluster-typed" % V.akka,
+
+    "com.typesafe.akka" %% "akka-discovery" % V.akka,
+    "com.typesafe.akka" %% "akka-cluster-tools" % V.akka
+  )
+
+  lazy val akkaHttp: Seq[ModuleID] = Seq(
+    "com.typesafe.akka" %% "akka-http" % V.akkaHttp
+  )
+
+  lazy val cats: Seq[ModuleID] = Seq(
+    "org.typelevel" %% "cats-core" % V.cats
+  )
+
+  lazy val monocle: Seq[ModuleID] = Seq(
+    "com.github.julien-truffaut" %% "monocle-core" % V.monocle,
+    "com.github.julien-truffaut" %% "monocle-macro" % V.monocle,
+    "com.github.julien-truffaut" %% "monocle-law" % V.monocle % "test"
+  )
+
+  lazy val configurationLibs: Seq[ModuleID] = Seq(
+    "com.typesafe" % "config" % V.typesafeConfig,
+    "com.github.pureconfig" %% "pureconfig" % V.pureConfig
+  )
+
+  lazy val scalaUri: Seq[ModuleID] = Seq(
+    "io.lemonlabs" %% "scala-uri" % V.scalaUri
+  )
+
+  lazy val decline: Seq[ModuleID] = Seq(
+    "com.monovore" %% "decline" % V.decline
+  )
+
+  lazy val jsoup: Seq[ModuleID] = Seq(
+    "org.jsoup" % "jsoup" % V.jsoup
+  )
+
+  lazy val circe: Seq[ModuleID] = Seq(
+    "io.circe" %% "circe-core" % V.circe,
+    "io.circe" %% "circe-generic" % V.circe,
+    "io.circe" %% "circe-parser" % V.circe
+  )
+
+  lazy val akkaHttpCirce: Seq[ModuleID] = Seq(
+    "de.heikoseeberger" %% "akka-http-circe" % V.akkaHttpCirce
+  )
+
+  lazy val logging: Seq[ModuleID] = Seq(
+    "ch.qos.logback" % "logback-classic" % V.logbackClassic,
+    "com.typesafe.scala-logging" %% "scala-logging" % V.scalaLogging,
+    "com.typesafe.akka" %% "akka-slf4j" % V.akkaSlf4j
+  )
+
+  lazy val scalaTest: Seq[ModuleID] = Seq(
+    "org.scalatest" %% "scalatest" % V.scalaTest % "test",
+    "org.scalatest" %% "scalatest-flatspec" % V.scalaTest % "test",
+    "org.scalatest" %% "scalatest-shouldmatchers" % V.scalaTest % "test"
+  )
+
+  lazy val javaAgentsLibs: Seq[ModuleID] = Seq(
+    "io.kamon" % "kanela-agent" % "1.0.5"
+  )
+}

project/Settings.scala

@@ -0,0 +1,35 @@
+import sbt._
+import sbt.Keys._
+import sbt.nio.Keys._
+
+object Settings {
+  lazy val sharedSettings = Seq(
+    Compile / packageDoc / mappings := Seq(),
+    Global / onChangedBuildSource := ReloadOnSourceChanges,
+
+    scalaVersion := "2.13.2",
+    organization := "com.pinkstack.realestate",
+
+    Test / fork := true,
+    Test / javaOptions ++= Seq("-Xmx2g"),
+
+    scalacOptions ++= Seq(
+      "-deprecation",
+      "-encoding", "UTF-8",
+      "-feature",
+      "-explaintypes",
+      "-unchecked",
+      "-Xlint:-unused,_",
+      "-language:implicitConversions",
+      "-language:higherKinds",
+      "-language:postfixOps",
+      "-Yrangepos",
+      "-target:jvm-1.11"
+    ),
+    resolvers ++= Seq(
+      Resolver.bintrayRepo("ovotech", "maven"),
+      "Confluent Maven Repository" at "https://packages.confluent.io/maven/",
+      "jitpack" at "https://jitpack.io"
+    )
+  )
+}

project/plugins.sbt

@@ -1,2 +1,15 @@
+logLevel := Level.Warn
+
+// addSbtPlugin("io.spray" % "sbt-revolver" % "0.9.1")
+
 addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.7.3")
+
 addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.10")
+
+addSbtPlugin("com.lightbend.sbt" % "sbt-javaagent" % "0.1.5")
+
+addSbtPlugin("io.kamon" % "sbt-kanela-runner" % "2.0.6")
+
+resolvers += Resolver.sonatypeRepo("releases")
+
+

src/main/resources/application.conf → scraper/src/main/resources/application.conf

@@ -11,17 +11,17 @@ pagination {
 }
 
 fetching {
-  categories-parallelism = 4
-  estates-parallelism = 4
+  categories-parallelism = 3
+  estates-parallelism = 3
 }
 
 throttling-estates = {
   # Number of elements "per"
-  elements = 20
+  elements = 10
 
 
   # Duration "per second"
   per = 1
 
-  maximum-burst = 30
+  maximum-burst = 20
 }