From 3e54b2f847279860c19bacb14dc32f59ef8282cd Mon Sep 17 00:00:00 2001
From: Oto Brglez <otobrglez@gmail.com>
Date: Fri, 3 Jul 2020 16:06:56 +0200
Subject: [PATCH 1/4] Experimenting with clustering.

---
 build.sbt                                     |  7 +++
 .../com/pinkstack/realestate/NepClient.scala  |  2 +-
 .../realestate/clustering/MainApp.scala       | 32 ++++++++++
 .../realestate/clustering/MainApp2.scala      | 46 +++++++++++++++
 .../realestate/clustering/MainApp3.scala      | 58 +++++++++++++++++++
 .../realestate/clustering/MainApp4.scala      | 45 ++++++++++++++
 6 files changed, 189 insertions(+), 1 deletion(-)
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala

diff --git a/build.sbt b/build.sbt
index 9ab08a9..ab027fb 100644
--- a/build.sbt
+++ b/build.sbt
@@ -6,9 +6,16 @@ scalaVersion := "2.13.2"
 
 libraryDependencies ++= Seq(
   // Akka and Akka Streams
+  "com.typesafe.akka" %% "akka-actor-typed" % "2.6.6",
   "com.typesafe.akka" %% "akka-stream" % "2.6.6",
   "com.typesafe.akka" %% "akka-http" % "10.1.12",
 
+  // Clustering and remoting
+  "com.typesafe.akka" %% "akka-cluster" % "2.6.6",
+  "com.typesafe.akka" %% "akka-cluster-typed" % "2.6.6",
+  "com.lightbend.akka.management" %% "akka-management-cluster-bootstrap" % "1.0.8",
+  "com.typesafe.akka" %% "akka-discovery" % "2.6.6",
+
   // FP
   "org.typelevel" %% "cats-core" % "2.0.0",
 
diff --git a/src/main/scala/com/pinkstack/realestate/NepClient.scala b/src/main/scala/com/pinkstack/realestate/NepClient.scala
index 59596d4..1d8a268 100644
--- a/src/main/scala/com/pinkstack/realestate/NepClient.scala
+++ b/src/main/scala/com/pinkstack/realestate/NepClient.scala
@@ -78,7 +78,7 @@ final case class NepClient(timeout: FiniteDuration = 4.seconds)(
       NepEstateParser.parse(document) match {
         case Valid(value: Estate) =>
           Some(value.copy(scrapedFromUrl = estateRequest.request.uri.toString))
-        case Invalid(errors: NonEmptyChain[ParserValidation]) =>
+        case Invalid(errors) =>
           logger.error(s"${errors} at ${estateRequest.request.uri}")
           None
       }
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
new file mode 100644
index 0000000..dc2cb81
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
@@ -0,0 +1,32 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
+import akka.cluster.Cluster
+import akka.cluster.ClusterEvent.MemberEvent
+import com.typesafe.scalalogging.LazyLogging
+
+class ExampleClusterListener extends Actor with ActorLogging {
+  val cluster = Cluster(context.system)
+
+  override def receive: Receive = {
+    case event: MemberEvent =>
+      log.info(s"Received ${event}")
+  }
+}
+
+class PingActor extends Actor with ActorLogging {
+  override def receive: Receive = {
+    case s: String =>
+      log.info(s"Received $s in ${self}")
+  }
+}
+
+object MainApp extends App with LazyLogging {
+  implicit val system = ActorSystem("estateApp")
+  implicit val context = system.dispatcher
+
+  val pingA = system.actorOf(Props[PingActor])
+  pingA ! "Hello Oto"
+
+  println("This is the MainApp")
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
new file mode 100644
index 0000000..19050e9
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
@@ -0,0 +1,46 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor.typed.scaladsl._
+import akka.actor.typed.ActorSystem
+import akka.actor.typed.Behavior
+import akka.cluster.typed.Cluster
+import akka.cluster.typed.Subscribe
+import com.typesafe.config.ConfigFactory
+import com.typesafe.scalalogging.LazyLogging
+
+object ClusterListener {
+
+  sealed trait Event
+
+  def apply(): Behavior[Event] = Behaviors.setup { ctx =>
+    Behaviors.receiveMessage {
+      case _ =>
+        println("ok")
+        Behaviors.same
+    }
+  }
+}
+
+object MainApp2 extends LazyLogging {
+
+  object RootBehavior {
+    def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] { context =>
+      context.spawn(ClusterListener(), "ClusterListener")
+      Behaviors.empty
+    }
+  }
+
+  def main(args: Array[String]): Unit = {
+    val ports: Seq[Int] = Seq(33200, 33201, 0)
+    ports.foreach(boot)
+  }
+
+  def boot(port: Int): Unit = {
+    val config = ConfigFactory.parseString(
+      s"""
+         |akka.remote.artery.canonical.host=0.0.0.0
+         |akka.remote.artery.canonical.port=$port""".stripMargin).withFallback(ConfigFactory.load)
+    ActorSystem[Nothing](RootBehavior(), s"ClusterSystem-$port", config)
+  }
+}
+
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
new file mode 100644
index 0000000..1593c38
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
@@ -0,0 +1,58 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor.{ActorSystem, CoordinatedShutdown}
+import akka.discovery.{Discovery, Lookup}
+import com.typesafe.config.ConfigFactory
+import com.typesafe.scalalogging.LazyLogging
+import scala.concurrent.duration._
+
+object MainApp3 extends LazyLogging {
+  def main(args: Array[String]): Unit = {
+    val config = ConfigFactory.parseString(
+      s"""
+         |akka.discovery.method = aggregate
+         |akka.discovery.aggregate {
+         |  discovery-methods = ["akka-dns", "config"]
+         |}
+         |akka.discovery.config.services = {
+         |  seeder = {
+         |    endpoints = [
+         |      {
+         |        host = 0.0.0.0
+         |        port = 6000
+         |      }
+         |   ]
+         |  },
+         |  fetchers = {
+         |    endpoints = [
+         |      {
+         |        host = 0.0.0.0
+         |        port = 6001
+         |      },
+         |      {
+         |        host = 0.0.0.0
+         |        port = 6002
+         |      }
+         |    ]
+         |  }
+         |}
+         |akka.cluster.shutdown-after-unsuccessful-join-seed-nodes = 10s
+         |akka.coordinated-shutdown.exit-jvm = on
+         |""".stripMargin).withFallback(ConfigFactory.load)
+
+    implicit val system = ActorSystem("main3", config)
+    implicit val executionContext = system.dispatcher
+
+    val discovery = Discovery(system).discovery
+
+    discovery.lookup(Lookup("fetchers"), 3.seconds).map { a =>
+      println(a.getAddresses)
+    }
+
+    CoordinatedShutdown(system).addJvmShutdownHook {
+      logger.info("Terminating...")
+      // clean shutdown of the flow
+      logger.info("Terminated... Bye")
+    }
+  }
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala
new file mode 100644
index 0000000..7839c84
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala
@@ -0,0 +1,45 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor.typed._
+import akka.actor.typed.scaladsl._
+
+object Greeter {
+
+  final case class Greet(whom: String)
+
+  final case class Greeted(whom: String, from: ActorRef[Greet])
+
+  def apply(): Behavior[Greet] = Behaviors.receive { (context, message) =>
+    context.log.info(s"Hello ${message.whom}")
+
+    if (message.whom.contains("xxx")) {
+      Behaviors.stopped
+    } else Behaviors.same
+  }
+}
+
+object MainApp4 {
+
+  final case class Hello(name: String)
+
+  def apply(): Behavior[Hello] = Behaviors.setup { context =>
+    context.log.info("Setup")
+
+    val greeter = context.spawn(Greeter(), "greeter")
+
+    Behaviors.receiveMessage { message =>
+      context.log.info(s"Got $message")
+
+      greeter ! Greeter.Greet(s"hello ${message.name}")
+      Behaviors.same
+    }
+  }
+
+  def main(args: Array[String]): Unit = {
+    val system: ActorSystem[MainApp4.Hello] = ActorSystem(MainApp4(), "mainSystem")
+
+    system ! Hello("Oto")
+    system ! Hello("Martina")
+    system ! Hello("xxx")
+  }
+}

From 50c043271b6be4e226dcb0a700eb2ba4f5fb6950 Mon Sep 17 00:00:00 2001
From: Oto Brglez <otobrglez@gmail.com>
Date: Mon, 13 Jul 2020 22:46:57 +0200
Subject: [PATCH 2/4] Experiments on clustering v1

---
 README.md                                     |  5 ++
 build.sbt                                     |  9 +-
 project/plugins.sbt                           |  5 ++
 src/main/resources/application.conf           |  8 +-
 src/main/resources/clusteringV4.conf          | 56 ++++++++++++
 src/main/resources/clusteringv3.conf          | 86 ++++++++++++++++++
 .../com/pinkstack/realestate/Domain.scala     |  6 +-
 .../pinkstack/realestate/EstateParser.scala   |  6 +-
 .../com/pinkstack/realestate/NepClient.scala  |  8 +-
 .../pinkstack/realestate/apps/MasterApp.scala |  5 --
 .../pinkstack/realestate/apps/Scraper.scala   | 13 ++-
 .../realestate/apps/ScraperStream.scala       | 30 -------
 .../realestate/clustering/ClusteringV2.scala  | 63 +++++++++++++
 .../clustering/StoreSystemApp.scala           | 87 ++++++++++++++++++
 .../clustering3/ClusterListener.scala         | 21 +++++
 .../clustering3/ClusterManager.scala          | 36 ++++++++
 .../realestate/clustering3/Domain.scala       | 12 +++
 .../realestate/clustering3/Node.scala         | 33 +++++++
 .../realestate/clustering3/ServerApp.scala    | 88 +++++++++++++++++++
 .../realestate/clustering4/NodeApp.scala      | 53 +++++++++++
 20 files changed, 578 insertions(+), 52 deletions(-)
 create mode 100644 src/main/resources/clusteringV4.conf
 create mode 100644 src/main/resources/clusteringv3.conf
 delete mode 100644 src/main/scala/com/pinkstack/realestate/apps/MasterApp.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/apps/ScraperStream.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
 create mode 100644 src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala

diff --git a/README.md b/README.md
index 556df64..0ebafdd 100644
--- a/README.md
+++ b/README.md
@@ -43,6 +43,11 @@ 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)
diff --git a/build.sbt b/build.sbt
index ab027fb..0f70235 100644
--- a/build.sbt
+++ b/build.sbt
@@ -6,15 +6,17 @@ scalaVersion := "2.13.2"
 
 libraryDependencies ++= Seq(
   // Akka and Akka Streams
+  "com.typesafe.akka" %% "akka-http" % "10.1.12",
   "com.typesafe.akka" %% "akka-actor-typed" % "2.6.6",
   "com.typesafe.akka" %% "akka-stream" % "2.6.6",
-  "com.typesafe.akka" %% "akka-http" % "10.1.12",
 
   // Clustering and remoting
   "com.typesafe.akka" %% "akka-cluster" % "2.6.6",
   "com.typesafe.akka" %% "akka-cluster-typed" % "2.6.6",
-  "com.lightbend.akka.management" %% "akka-management-cluster-bootstrap" % "1.0.8",
+  // "com.lightbend.akka.management" %% "akka-management" % "1.0.8",
+  // "com.lightbend.akka.management" %% "akka-management-cluster-bootstrap" % "1.0.8",
   "com.typesafe.akka" %% "akka-discovery" % "2.6.6",
+  "com.typesafe.akka" %% "akka-cluster-tools" % "2.6.6",
 
   // FP
   "org.typelevel" %% "cats-core" % "2.0.0",
@@ -42,6 +44,9 @@ libraryDependencies ++= Seq(
   "io.circe" %% "circe-generic" % "0.12.3",
   "io.circe" %% "circe-parser" % "0.12.3",
 
+  // JSON + Akka
+  "de.heikoseeberger" %% "akka-http-circe" % "1.31.0",
+
   // Logging
   "ch.qos.logback" % "logback-classic" % "1.2.3",
   "com.typesafe.scala-logging" %% "scala-logging" % "3.9.2",
diff --git a/project/plugins.sbt b/project/plugins.sbt
index bff0726..a840e45 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -1,2 +1,7 @@
+logLevel := Level.Warn
+
 addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.7.3")
+
 addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.10")
+
+resolvers += Resolver.sonatypeRepo("releases")
diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf
index 0227d84..3febf6a 100644
--- a/src/main/resources/application.conf
+++ b/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
 }
diff --git a/src/main/resources/clusteringV4.conf b/src/main/resources/clusteringV4.conf
new file mode 100644
index 0000000..f02205e
--- /dev/null
+++ b/src/main/resources/clusteringV4.conf
@@ -0,0 +1,56 @@
+akka {
+  # Log the complete configuration at INFO level when the actor system is started.
+  # This is useful when you are uncertain of what configuration is used.
+  ## log-config-on-start = on
+
+  # Loggers to register at boot time (akka.event.Logging$DefaultLogger logs
+  # to STDOUT)
+  loggers = ["akka.event.slf4j.Slf4jLogger"]
+
+  # Log level used by the configured loggers (see "loggers") as soon
+  # as they have been started; before that, see "stdout-loglevel"
+  # Options: OFF, ERROR, WARNING, INFO, DEBUG
+  loglevel = "INFO" # "DEBUG"
+
+  # Log level for the very basic logger activated during ActorSystem startup.
+  # This logger prints the log messages to stdout (System.out).
+  # Options: OFF, ERROR, WARNING, INFO, DEBUG
+  stdout-loglevel = "DEBUG"
+
+  # Filter of log events that is used by the LoggingAdapter before
+  # publishing log events to the eventStream.
+  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
+
+  http {
+    client {
+      idle-timeout = 120s
+    }
+    host-connection-pool {
+      idle-timeout = 200s
+    }
+  }
+
+  actor {
+    provider = "cluster"
+  }
+
+  remote {
+    log-remote-lifecycle-events = off
+    artery.enabled = on
+    artery.canonical.hostname = "127.0.0.1"
+    artery.canonical.port = 0
+  }
+
+  cluster {
+    seed-nodes = [
+      "akka://AppV4@127.0.0.1:4000",
+      "akka://AppV4@127.0.0.1:4001",
+    ]
+    auto-down-unreachable-after = 10s
+    downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
+  }
+
+  coordinated-shutdown.exit-jvm = on
+  coordinated-shutdown.terminate-actor-system = on
+  cluster.shutdown-after-unsuccessful-join-seed-nodes = 20s
+}
\ No newline at end of file
diff --git a/src/main/resources/clusteringv3.conf b/src/main/resources/clusteringv3.conf
new file mode 100644
index 0000000..922820b
--- /dev/null
+++ b/src/main/resources/clusteringv3.conf
@@ -0,0 +1,86 @@
+akka {
+  # Log the complete configuration at INFO level when the actor system is started.
+  # This is useful when you are uncertain of what configuration is used.
+  ## log-config-on-start = on
+
+  # Loggers to register at boot time (akka.event.Logging$DefaultLogger logs
+  # to STDOUT)
+  loggers = ["akka.event.slf4j.Slf4jLogger"]
+
+  # Log level used by the configured loggers (see "loggers") as soon
+  # as they have been started; before that, see "stdout-loglevel"
+  # Options: OFF, ERROR, WARNING, INFO, DEBUG
+  loglevel = "DEBUG" # "DEBUG"
+
+  # Log level for the very basic logger activated during ActorSystem startup.
+  # This logger prints the log messages to stdout (System.out).
+  # Options: OFF, ERROR, WARNING, INFO, DEBUG
+  stdout-loglevel = "DEBUG"
+
+  # Filter of log events that is used by the LoggingAdapter before
+  # publishing log events to the eventStream.
+  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
+
+  http {
+    client {
+      idle-timeout = 120s
+    }
+    host-connection-pool {
+      idle-timeout = 200s
+    }
+  }
+
+  actor {
+    provider = "cluster"
+  }
+
+  remote {
+    log-remote-lifecycle-events = off
+    artery.enabled = on
+
+    artery {
+      canonical.hostname = "localhost"
+      canonical.port = 0
+      canonical.port = ${?C_PORT}
+    }
+  }
+
+  cluster {
+    seed-nodes = [
+      "akka://"${clustering.cluster.name}"@"${clustering.seed-ip}":"${clustering.seed-port},
+    ]
+    auto-down-unreachable-after = 10s
+    downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
+  }
+
+  coordinated-shutdown.exit-jvm = on
+  coordinated-shutdown.terminate-actor-system = on
+
+  cluster.shutdown-after-unsuccessful-join-seed-nodes = 20s
+}
+
+http {
+  ip = 127.0.0.1
+  ip = ${?SERVER_IP}
+  port = 7000
+  port = ${?SERVER_PORT}
+}
+
+clustering {
+  ip = "127.0.0.1"
+  ip = ${?CLUSTER_IP}
+
+  port = 4000
+  port = ${?CLUSTER_PORT}
+
+  seed-ip = "localhost"
+  seed-ip = ${?CLUSTER_SEED_IP}
+
+  seed-port = 4000
+  seed-port = ${?CLUSTER_SEED_PORT}
+
+  seed-port_two = 4001
+  seed-port_two = ${?CLUSTER_SEED_PORT_TWO}
+
+  cluster.name = "mySystem"
+}
\ No newline at end of file
diff --git a/src/main/scala/com/pinkstack/realestate/Domain.scala b/src/main/scala/com/pinkstack/realestate/Domain.scala
index 442ebac..ce8a7ce 100644
--- a/src/main/scala/com/pinkstack/realestate/Domain.scala
+++ b/src/main/scala/com/pinkstack/realestate/Domain.scala
@@ -16,8 +16,6 @@ object Domain {
       s"EstateRequest ${request.uri}"
   }
 
-  implicit val lemonLabsUrlToAkkaUrl: Url => Uri = url => Uri(url.toString())
-
   case class Location(latitude: Double, longitude: Double)
 
   case class Estate(title: String,
@@ -29,3 +27,7 @@ object Domain {
                     categories: Map[String, String] = Map.empty[String, String],
                     location: Option[Location] = None)
 }
+
+object Implicits {
+  implicit val lemonLabsUrlToAkkaUrl: Url => Uri = url => Uri(url.toString())
+}
\ No newline at end of file
diff --git a/src/main/scala/com/pinkstack/realestate/EstateParser.scala b/src/main/scala/com/pinkstack/realestate/EstateParser.scala
index 45ca9f2..e2bf868 100644
--- a/src/main/scala/com/pinkstack/realestate/EstateParser.scala
+++ b/src/main/scala/com/pinkstack/realestate/EstateParser.scala
@@ -65,17 +65,15 @@ object NepEstateParser {
     categoriesOpt.fold[ValidationResult[Map[String, String]]](NoCategories.invalidNec)(_.validNec)
   }
 
-  val validatedPrice: Transformer[Double] = { document =>
+  val validatedPrice: Transformer[Double] = document =>
     Option(document.selectFirst("meta[itemprop='price']"))
       .flatMap(_.attr("content").toDoubleOption)
       .fold[ValidationResult[Double]](NoPrice.invalidNec)(_.validNec)
-  }
 
-  val validatedRefNumber: Transformer[String] = { document =>
+  val validatedRefNumber: Transformer[String] = document =>
     Option(document.selectFirst("span.ikona-sh"))
       .map(_.attr("data-id"))
       .fold[ValidationResult[String]](NoRefNumber.invalidNec)(_.validNec)
-  }
 
   val validatedLocationDetails: Transformer[Map[String, Vector[String]]] = { document =>
     import scala.jdk.CollectionConverters._
diff --git a/src/main/scala/com/pinkstack/realestate/NepClient.scala b/src/main/scala/com/pinkstack/realestate/NepClient.scala
index 1d8a268..9b7c8df 100644
--- a/src/main/scala/com/pinkstack/realestate/NepClient.scala
+++ b/src/main/scala/com/pinkstack/realestate/NepClient.scala
@@ -1,12 +1,14 @@
 package com.pinkstack.realestate
 
+import cats._
+import cats.implicits._
+import cats.data.NonEmptyList
 import akka.NotUsed
 import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.HttpRequest
 import akka.stream.ThrottleMode
 import akka.stream.scaladsl.{Flow, Merge, Source}
-import cats.data.NonEmptyChain
 import cats.data.Validated.{Invalid, Valid}
 import com.typesafe.scalalogging.LazyLogging
 import io.lemonlabs.uri.typesafe.dsl._
@@ -23,6 +25,7 @@ final case class NepClient(timeout: FiniteDuration = 4.seconds)(
   val configuration: Configuration) extends LazyLogging {
 
   import Domain._
+  import Implicits._
 
   final val ROOT_URL = Url.parse("https://www.nepremicnine.net")
 
@@ -73,6 +76,9 @@ final case class NepClient(timeout: FiniteDuration = 4.seconds)(
       .map(doc => (categoryRequests(doc), estateRequests(doc)))
   }
 
+  val fetchEstatePageViaUrl: Url => Future[Option[Estate]] = url =>
+    fetchEstatePage(EstateRequest(HttpRequest(uri = url), ""))
+
   val fetchEstatePage: EstateRequest => Future[Option[Estate]] = { estateRequest =>
     val parse: Document => Option[Estate] = { document =>
       NepEstateParser.parse(document) match {
diff --git a/src/main/scala/com/pinkstack/realestate/apps/MasterApp.scala b/src/main/scala/com/pinkstack/realestate/apps/MasterApp.scala
deleted file mode 100644
index e4d18ce..0000000
--- a/src/main/scala/com/pinkstack/realestate/apps/MasterApp.scala
+++ /dev/null
@@ -1,5 +0,0 @@
-package com.pinkstack.realestate.apps
-
-object MasterApp extends App {
-
-}
diff --git a/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala b/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala
index 8c9d07f..d728670 100644
--- a/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala
+++ b/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala
@@ -30,10 +30,18 @@ object Implicits {
   }
 }
 
+import Implicits._
+
 object Main {
 
   import Implicits._
 
+  val estatePrinter: Option[Estate] => Unit = {
+    case Some(estate: Estate) =>
+      println(estate.asJson.noSpaces)
+    case None =>
+  }
+
   def withConfiguration(categories: CategoryList, numberOfPages: Int): Unit = {
     val configurationLens: Configuration => Configuration =
       GenLens[Configuration](_.seed.initialCategories).set(categories) compose
@@ -45,14 +53,11 @@ object Main {
 
     NepClient()
       .pipeline
-      .collect { case Some(estate: Estate) => estate.asJson.noSpaces }
-      .runWith(Sink.foreach(println))
+      .runWith(Sink.foreach(estatePrinter))
       .onComplete(_ => system.terminate())
   }
 }
 
-import com.pinkstack.realestate.apps.Implicits._
-
 object Scraper extends CommandApp(name = "scraper",
   header = "Scraper is used for fetching real estate listings directly in CLI.",
   main = {
diff --git a/src/main/scala/com/pinkstack/realestate/apps/ScraperStream.scala b/src/main/scala/com/pinkstack/realestate/apps/ScraperStream.scala
deleted file mode 100644
index 90d7c56..0000000
--- a/src/main/scala/com/pinkstack/realestate/apps/ScraperStream.scala
+++ /dev/null
@@ -1,30 +0,0 @@
-package com.pinkstack.realestate.apps
-
-import akka.actor.ActorSystem
-import akka.stream.scaladsl._
-import com.pinkstack.realestate.Domain._
-import com.pinkstack.realestate.{Configuration, NepClient}
-import com.typesafe.scalalogging.LazyLogging
-import io.circe.generic.auto._
-import io.circe.syntax._
-
-import scala.concurrent.ExecutionContextExecutor
-
-object ScraperStream extends App with LazyLogging {
-  logger.info("ScraperApp starting.")
-
-  implicit val system: ActorSystem = ActorSystem("scraperAppTwo")
-  implicit val context: ExecutionContextExecutor = system.dispatcher
-  implicit val configuration: Configuration = Configuration.loadOrThrow
-
-  val estatePrinter: Option[Estate] => Unit = {
-    case Some(estate: Estate) =>
-      println(estate.asJson.noSpaces)
-    case None =>
-  }
-
-  NepClient()
-    .pipeline
-    .runWith(Sink.foreach(estatePrinter))
-    .onComplete(_ => system.terminate())
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala b/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
new file mode 100644
index 0000000..fe10bae
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
@@ -0,0 +1,63 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor._
+import akka.cluster.Cluster
+import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, MemberRemoved, MemberUp, UnreachableMember}
+import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.config.ConfigFactory
+
+class ClusterListenerActor extends Actor with ActorLogging {
+  val cluster = Cluster(context.system)
+
+  override def preStart(): Unit = {
+    cluster.subscribe(self,
+      initialStateMode = InitialStateAsEvents,
+      classOf[MemberEvent], classOf[UnreachableMember]
+    )
+  }
+
+  override def postStop(): Unit = {
+    cluster.unsubscribe(self)
+  }
+
+  override def receive: Receive = {
+    case MemberUp(member) =>
+      log.info("Member is up {}", member.address)
+    case UnreachableMember(member) =>
+      log.info("Member detected as unreachable {}", member.address)
+    case MemberRemoved(member, previousStatus) =>
+      log.info("Member removed {} after {}", member.address, previousStatus)
+    case event: MemberEvent =>
+      log.info("Got event {}", event)
+  }
+}
+
+object ClusteringV2 extends LazyLogging {
+  def main(args: Array[String]): Unit = {
+    logger.info("Clustering demo")
+
+    val clusteringConfig = ConfigFactory.load(
+      """
+        |akka {
+        | actor {
+        |   provider = "cluster"
+        | }
+        | remote.artery {
+        |    canonical {
+        |      hostname = "0.0.0.0"
+        |      port = 2551
+        |    }
+        | }
+        | cluster {
+        |    seed-nodes = [
+        |      "akka://ClusterSystem@127.0.0.1:2551",
+        |      "akka://ClusterSystem@127.0.0.1:2552"]
+        |
+        |   downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
+        | }
+        |}
+        |""".stripMargin)
+
+    println(clusteringConfig)
+  }
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala b/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
new file mode 100644
index 0000000..660eb62
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
@@ -0,0 +1,87 @@
+package com.pinkstack.realestate.clustering
+
+import akka.actor.typed._
+import akka.actor.typed.scaladsl._
+import com.typesafe.scalalogging.LazyLogging
+
+object Domain {
+
+  sealed trait StoreMessage
+
+  case class AddItem(item: String) extends StoreMessage
+
+  case class RemoveItem(item: String) extends StoreMessage
+
+  case object ValidateCart extends StoreMessage
+
+}
+
+object StoreSystemApp extends LazyLogging {
+
+  import Domain._
+
+  def main(args: Array[String]): Unit = {
+    logger.info("StoreSystemApp booting,...")
+
+    val storeRootActor = ActorSystem(
+      Behaviors.receiveMessage[StoreMessage] { message =>
+        message match {
+          case AddItem(item) => println(s"Adding item to cart ${item}")
+          case RemoveItem(item) => println(s"Removing item ${item}")
+          case ValidateCart => println("cart is good")
+        }
+        Behaviors.same
+      }, "storeRootActor"
+    )
+
+    storeRootActor ! AddItem("cake")
+    storeRootActor ! AddItem("beer")
+
+
+    val storeRootActorMutable = ActorSystem(
+      Behaviors.setup[StoreMessage] { ctx =>
+        // Local state
+        var items: Set[String] = Set()
+
+        Behaviors.receiveMessage[StoreMessage] { message =>
+          message match {
+            case AddItem(item) =>
+              items += item
+              println(s"Item ${item} added.")
+            case RemoveItem(item) =>
+              items -= item
+              println(s"Removed item ${item}")
+            case ValidateCart => println("cart is good")
+          }
+          Behaviors.same
+        }
+      }
+      , "storeRootActorMutalbe"
+    )
+
+    def shoppingBehaviour(items: Set[String]): Behavior[StoreMessage] =
+      Behaviors.receiveMessage[StoreMessage] {
+        case AddItem(item) =>
+          println("adding")
+          shoppingBehaviour(items + item)
+        case RemoveItem(item) =>
+          println("removing")
+          shoppingBehaviour(items - item)
+        case ValidateCart =>
+          Behaviors.same
+      }
+
+    storeRootActorMutable ! AddItem("coke")
+    storeRootActorMutable ! AddItem("coke")
+    storeRootActorMutable ! AddItem("beer")
+
+    val rootOnlineActor = ActorSystem(
+      Behaviors.setup[StoreMessage] { ctx =>
+        ctx.spawn(shoppingBehaviour(Set()), "otoCart")
+        Behaviors.empty
+      }, "onlineStore"
+    )
+
+    rootOnlineActor ! AddItem("xxx")
+  }
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
new file mode 100644
index 0000000..76cfc6a
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
@@ -0,0 +1,21 @@
+package com.pinkstack.realestate.clustering3
+
+import akka.actor.{Actor, ActorLogging}
+import akka.cluster.Cluster
+import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, UnreachableMember}
+import com.pinkstack.realestate.clustering3.Domain.NodeID
+
+class ClusterListener(nodeID: NodeID, cluster: Cluster) extends Actor with ActorLogging {
+  override def preStart(): Unit =
+    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
+      classOf[MemberEvent], classOf[UnreachableMember])
+
+  override def postStop(): Unit = cluster.unsubscribe(self)
+
+  override def receive: Receive = {
+    case x: Any =>
+      println("~~~>")
+      println(x)
+      println("~~~>")
+  }
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
new file mode 100644
index 0000000..107ed38
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
@@ -0,0 +1,36 @@
+package com.pinkstack.realestate.clustering3
+
+import akka.actor.{Actor, ActorLogging, ActorRef, Props}
+import akka.cluster.{Cluster, Member}
+import com.pinkstack.realestate.clustering3.Domain.NodeID
+
+class ClusterManager(nodeID: NodeID) extends Actor with ActorLogging {
+  val cluster: Cluster = Cluster(context.system)
+  val listener: ActorRef = context.actorOf(Props(classOf[ClusterListener], nodeID, cluster), "clusterListener")
+
+  import ClusterManager._
+
+  override def receive: Receive = {
+    case GetMembers =>
+      sender ! MembersCollected {
+        cluster.state.members.toList.map { member: Member =>
+          Map[String, String](
+            "address" -> member.address.toString,
+            "status" -> member.status.toString
+          )
+        }
+      }
+  }
+}
+
+object ClusterManager {
+
+  sealed trait ClusterMessage
+
+  sealed trait ClusterEvent
+
+  case object GetMembers extends ClusterMessage
+
+  case class MembersCollected(members: List[Map[String, String]]) extends ClusterEvent
+
+}
\ No newline at end of file
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala b/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
new file mode 100644
index 0000000..3f52d8a
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
@@ -0,0 +1,12 @@
+package com.pinkstack.realestate.clustering3
+
+object Domain {
+  type NodeID = String
+
+  sealed trait Command
+
+  case class FetchEstate(url: String) extends Command
+
+  case object ScrapeAll extends Command
+
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala b/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
new file mode 100644
index 0000000..b94cef4
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
@@ -0,0 +1,33 @@
+package com.pinkstack.realestate.clustering3
+
+import akka.actor.{Actor, ActorLogging, ActorRef, Props}
+import com.pinkstack.realestate.clustering3.ClusterManager.GetMembers
+import com.pinkstack.realestate.clustering3.Domain.{FetchEstate, NodeID, ScrapeAll}
+import com.pinkstack.realestate.{Configuration, NepClient}
+import io.lemonlabs.uri.Url
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+import akka.pattern.{ask, pipe}
+
+class Node(nodeID: NodeID) extends Actor with ActorLogging {
+  val configuration: Configuration = Configuration.loadOrThrow
+  val nepClient: NepClient = NepClient(4.seconds)(this.context.system, this.context.system.dispatcher, configuration)
+  implicit val executionContext: ExecutionContext = this.context.dispatcher
+
+  val clusterManager: ActorRef = context.actorOf(Props(classOf[ClusterManager], nodeID), "clusterManager")
+
+  override def preStart(): Unit = log.info("preStart of {} in {}", nodeID, self)
+
+  override def receive: Receive = {
+    case FetchEstate(url) =>
+      log.info("Fetching {}", url)
+      nepClient.fetchEstatePageViaUrl(Url.parse(url)) pipeTo sender
+    case command@GetMembers =>
+      ((clusterManager ? command) (3.seconds)) pipeTo sender
+    case ScrapeAll =>
+
+    case x: Any =>
+      log.info("Got {} at {}", x, self)
+  }
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
new file mode 100644
index 0000000..7a3f47d
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
@@ -0,0 +1,88 @@
+package com.pinkstack.realestate.clustering3
+
+import akka.actor._
+import com.typesafe.config.{Config, ConfigFactory}
+
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.concurrent.duration._
+import akka.http.scaladsl._
+import akka.http.scaladsl.model.{HttpRequest, StatusCodes, Uri}
+import akka.http.scaladsl.server.Route
+import akka.http.scaladsl.server.Directives._
+import com.pinkstack.realestate.clustering3.ClusterManager.{GetMembers, MembersCollected}
+import com.typesafe.scalalogging.LazyLogging
+import akka.pattern.ask
+import akka.pattern.pipe
+import akka.pattern.PipeToSupport
+import akka.util.Timeout
+import com.pinkstack.realestate.Domain.{Estate, EstateRequest}
+import com.pinkstack.realestate.{Configuration, NepClient}
+import io.lemonlabs.uri.Url
+
+import Domain._
+
+
+trait ServerRoutes {
+  this: LazyLogging =>
+
+  import de.heikoseeberger.akkahttpcirce.FailFastCirceSupport._
+  import io.circe.generic.auto._
+
+  implicit val system: ActorSystem
+  implicit val dispatcher: ExecutionContext
+
+  def stopSystem: Future[String] = for {
+    _ <- system.terminate()
+    _ = logger.info("System terminated.")
+  } yield "Terminated"
+
+  val fetchEstate: (ActorRef, String) => Future[Option[Estate]] = (nodeActor, url) =>
+    (nodeActor ? FetchEstate(url)) (3.seconds).mapTo[Option[Estate]]
+
+  val getMembers: ActorRef => Future[MembersCollected] = { nodeActorRef =>
+    (nodeActorRef ? GetMembers) (3.seconds).mapTo[MembersCollected]
+  }
+  val scrapeAll: ActorRef => Future[Map[String, String]] = { nodeActorRef =>
+    nodeActorRef ! ScrapeAll
+    Future.successful(Map[String, String]("scraping" -> "started,..."))
+  }
+
+  def basicRoutes(nodeActorRef: ActorRef): Route = {
+    concat(
+      pathSingleSlash(get(complete(StatusCodes.OK, "ok"))),
+      path("stop")(get(complete(stopSystem))),
+      path("fetchEstate") {
+        parameters(Symbol("url").as[String]) { (url: String) =>
+          onSuccess(fetchEstate(nodeActorRef, url)) {
+            case Some(estate) => complete(StatusCodes.OK, estate)
+            case None => complete(StatusCodes.BadRequest, Map("error" -> "Sorry, fetching failed."))
+          }
+        }
+      },
+      path("getMembers")(get {
+        onSuccess(getMembers(nodeActorRef)) { result: MembersCollected =>
+          complete(Map("members" -> result.members))
+        }
+      }),
+      path("scrape")(get(complete(scrapeAll(nodeActorRef)))
+      )
+    )
+  }
+}
+
+object ServerApp extends App with LazyLogging with ServerRoutes {
+  val config: Config = ConfigFactory.load("clusteringv3.conf")
+  implicit val system: ActorSystem = ActorSystem("mySystem", config)
+  implicit val dispatcher: ExecutionContext = system.dispatcher
+
+  val address: String = config.getString("http.ip")
+  val port: Int = config.getInt("http.port")
+
+  val node: ActorRef = system.actorOf(Props(classOf[Node], "x"), "node")
+
+  Http().bindAndHandle(basicRoutes(node), address, port)
+  logger.info(s"🚀 Node is listening on http://{}:{}", address, port)
+  logger.info(s"🛑 Stop it at http://{}:{}/stop", address, port)
+
+  Await.result(system.whenTerminated, Duration.Inf)
+}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala b/src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala
new file mode 100644
index 0000000..3e52dd8
--- /dev/null
+++ b/src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala
@@ -0,0 +1,53 @@
+/**
+ * Author: Oto Brglez - <otobrglez@gmail.com>
+ */
+package com.pinkstack.realestate.clustering4
+
+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)
+)
\ No newline at end of file

From a838d8453ac2b3035c044a789d113e91aa56fe32 Mon Sep 17 00:00:00 2001
From: Oto Brglez <otobrglez@gmail.com>
Date: Tue, 14 Jul 2020 00:21:02 +0200
Subject: [PATCH 3/4] Cleanup and refactoring.

---
 README.md                                     |   8 +-
 build.sbt                                     | 105 ++++++------------
 .../realestate/cluster}/NodeApp.scala         |   2 +-
 project/Dependencies.scala                    |  98 ++++++++++++++++
 project/Settings.scala                        |  35 ++++++
 project/plugins.sbt                           |   8 ++
 {src => scraper/src}/main/resources/akka.conf |   0
 .../src}/main/resources/application.conf      |   0
 .../src}/main/resources/clusteringV4.conf     |   0
 .../src}/main/resources/clusteringv3.conf     |   0
 .../src}/main/resources/logback.xml           |   2 +-
 .../src}/main/resources/quick.conf            |   0
 .../realestate/scraper}/Configuration.scala   |   2 +-
 .../realestate/scraper}/Domain.scala          |   2 +-
 .../realestate/scraper}/EstateParser.scala    |   4 +-
 .../realestate/scraper}/NepClient.scala       |   2 +-
 .../realestate/scraper}/apps/Scraper.scala    |   6 +-
 .../src}/test/resources/404-error-page.html   |   0
 .../test/resources/dravograd-stanovanje.html  |   0
 .../ljubljana-skladisce-no-price.html         |   0
 .../src}/test/resources/novigrad-hisa.html    |   0
 .../src}/test/resources/sela-posest.html      |   0
 .../src}/test/scala/NepEstateParserSpec.scala |  11 +-
 .../realestate/clustering/ClusteringV2.scala  |  63 -----------
 .../realestate/clustering/MainApp.scala       |  32 ------
 .../realestate/clustering/MainApp2.scala      |  46 --------
 .../realestate/clustering/MainApp3.scala      |  58 ----------
 .../realestate/clustering/MainApp4.scala      |  45 --------
 .../clustering/StoreSystemApp.scala           |  87 ---------------
 .../clustering3/ClusterListener.scala         |  21 ----
 .../clustering3/ClusterManager.scala          |  36 ------
 .../realestate/clustering3/Domain.scala       |  12 --
 .../realestate/clustering3/Node.scala         |  33 ------
 .../realestate/clustering3/ServerApp.scala    |  88 ---------------
 worksheets/coordinates.sc                     |  11 --
 worksheets/parsing.sc                         |  16 ---
 worksheets/ranges.sc                          |   1 -
 37 files changed, 194 insertions(+), 640 deletions(-)
 rename {src/main/scala/com/pinkstack/realestate/clustering4 => cluster/src/main/scala/com/pinkstack/realestate/cluster}/NodeApp.scala (97%)
 create mode 100644 project/Dependencies.scala
 create mode 100644 project/Settings.scala
 rename {src => scraper/src}/main/resources/akka.conf (100%)
 rename {src => scraper/src}/main/resources/application.conf (100%)
 rename {src => scraper/src}/main/resources/clusteringV4.conf (100%)
 rename {src => scraper/src}/main/resources/clusteringv3.conf (100%)
 rename {src => scraper/src}/main/resources/logback.xml (93%)
 rename {src => scraper/src}/main/resources/quick.conf (100%)
 rename {src/main/scala/com/pinkstack/realestate => scraper/src/main/scala/com/pinkstack/realestate/scraper}/Configuration.scala (93%)
 rename {src/main/scala/com/pinkstack/realestate => scraper/src/main/scala/com/pinkstack/realestate/scraper}/Domain.scala (96%)
 rename {src/main/scala/com/pinkstack/realestate => scraper/src/main/scala/com/pinkstack/realestate/scraper}/EstateParser.scala (98%)
 rename {src/main/scala/com/pinkstack/realestate => scraper/src/main/scala/com/pinkstack/realestate/scraper}/NepClient.scala (99%)
 rename {src/main/scala/com/pinkstack/realestate => scraper/src/main/scala/com/pinkstack/realestate/scraper}/apps/Scraper.scala (93%)
 rename {src => scraper/src}/test/resources/404-error-page.html (100%)
 rename {src => scraper/src}/test/resources/dravograd-stanovanje.html (100%)
 rename {src => scraper/src}/test/resources/ljubljana-skladisce-no-price.html (100%)
 rename {src => scraper/src}/test/resources/novigrad-hisa.html (100%)
 rename {src => scraper/src}/test/resources/sela-posest.html (100%)
 rename {src => scraper/src}/test/scala/NepEstateParserSpec.scala (88%)
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
 delete mode 100644 src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
 delete mode 100644 worksheets/coordinates.sc
 delete mode 100644 worksheets/parsing.sc
 delete mode 100644 worksheets/ranges.sc

diff --git a/README.md b/README.md
index 0ebafdd..44f28e4 100644
--- a/README.md
+++ b/README.md
@@ -17,14 +17,14 @@ java -jar target/*/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.
@@ -52,6 +52,6 @@ CATEGORY_PAGES_LIMIT=3
 
 - [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
diff --git a/build.sbt b/build.sbt
index 0f70235..d1a19f5 100644
--- a/build.sbt
+++ b/build.sbt
@@ -1,72 +1,33 @@
-name := "pinkstack-realestate"
-
-version := "0.0.1"
-
-scalaVersion := "2.13.2"
-
-libraryDependencies ++= Seq(
-  // Akka and Akka Streams
-  "com.typesafe.akka" %% "akka-http" % "10.1.12",
-  "com.typesafe.akka" %% "akka-actor-typed" % "2.6.6",
-  "com.typesafe.akka" %% "akka-stream" % "2.6.6",
-
-  // Clustering and remoting
-  "com.typesafe.akka" %% "akka-cluster" % "2.6.6",
-  "com.typesafe.akka" %% "akka-cluster-typed" % "2.6.6",
-  // "com.lightbend.akka.management" %% "akka-management" % "1.0.8",
-  // "com.lightbend.akka.management" %% "akka-management-cluster-bootstrap" % "1.0.8",
-  "com.typesafe.akka" %% "akka-discovery" % "2.6.6",
-  "com.typesafe.akka" %% "akka-cluster-tools" % "2.6.6",
-
-  // 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",
-
-  // JSON + Akka
-  "de.heikoseeberger" %% "akka-http-circe" % "1.31.0",
-
-  // 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"
\ No newline at end of file
+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)
diff --git a/src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala b/cluster/src/main/scala/com/pinkstack/realestate/cluster/NodeApp.scala
similarity index 97%
rename from src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala
rename to cluster/src/main/scala/com/pinkstack/realestate/cluster/NodeApp.scala
index 3e52dd8..6e02dee 100644
--- a/src/main/scala/com/pinkstack/realestate/clustering4/NodeApp.scala
+++ b/cluster/src/main/scala/com/pinkstack/realestate/cluster/NodeApp.scala
@@ -1,7 +1,7 @@
 /**
  * Author: Oto Brglez - <otobrglez@gmail.com>
  */
-package com.pinkstack.realestate.clustering4
+package com.pinkstack.realestate.cluster
 
 import akka.actor.typed._
 import akka.actor.typed.scaladsl._
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
new file mode 100644
index 0000000..861fd9d
--- /dev/null
+++ b/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"
+  )
+}
diff --git a/project/Settings.scala b/project/Settings.scala
new file mode 100644
index 0000000..757358b
--- /dev/null
+++ b/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"
+    )
+  )
+}
\ No newline at end of file
diff --git a/project/plugins.sbt b/project/plugins.sbt
index a840e45..3eca7d1 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -1,7 +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")
+
+
diff --git a/src/main/resources/akka.conf b/scraper/src/main/resources/akka.conf
similarity index 100%
rename from src/main/resources/akka.conf
rename to scraper/src/main/resources/akka.conf
diff --git a/src/main/resources/application.conf b/scraper/src/main/resources/application.conf
similarity index 100%
rename from src/main/resources/application.conf
rename to scraper/src/main/resources/application.conf
diff --git a/src/main/resources/clusteringV4.conf b/scraper/src/main/resources/clusteringV4.conf
similarity index 100%
rename from src/main/resources/clusteringV4.conf
rename to scraper/src/main/resources/clusteringV4.conf
diff --git a/src/main/resources/clusteringv3.conf b/scraper/src/main/resources/clusteringv3.conf
similarity index 100%
rename from src/main/resources/clusteringv3.conf
rename to scraper/src/main/resources/clusteringv3.conf
diff --git a/src/main/resources/logback.xml b/scraper/src/main/resources/logback.xml
similarity index 93%
rename from src/main/resources/logback.xml
rename to scraper/src/main/resources/logback.xml
index dd16d27..22dbfbc 100644
--- a/src/main/resources/logback.xml
+++ b/scraper/src/main/resources/logback.xml
@@ -18,7 +18,7 @@
 
     <logger name="com.zaxxer" level="INFO"/>
 
-    <logger name="com.pinkstack.realestate.NepClient" level="DEBUG" />
+    <logger name="com.pinkstack.realestate.scraper.NepClient" level="DEBUG" />
 
 
 
diff --git a/src/main/resources/quick.conf b/scraper/src/main/resources/quick.conf
similarity index 100%
rename from src/main/resources/quick.conf
rename to scraper/src/main/resources/quick.conf
diff --git a/src/main/scala/com/pinkstack/realestate/Configuration.scala b/scraper/src/main/scala/com/pinkstack/realestate/scraper/Configuration.scala
similarity index 93%
rename from src/main/scala/com/pinkstack/realestate/Configuration.scala
rename to scraper/src/main/scala/com/pinkstack/realestate/scraper/Configuration.scala
index 5a27c96..0417add 100644
--- a/src/main/scala/com/pinkstack/realestate/Configuration.scala
+++ b/scraper/src/main/scala/com/pinkstack/realestate/scraper/Configuration.scala
@@ -1,4 +1,4 @@
-package com.pinkstack.realestate
+package com.pinkstack.realestate.scraper
 
 import pureconfig._
 import pureconfig.generic.auto._
diff --git a/src/main/scala/com/pinkstack/realestate/Domain.scala b/scraper/src/main/scala/com/pinkstack/realestate/scraper/Domain.scala
similarity index 96%
rename from src/main/scala/com/pinkstack/realestate/Domain.scala
rename to scraper/src/main/scala/com/pinkstack/realestate/scraper/Domain.scala
index ce8a7ce..880f146 100644
--- a/src/main/scala/com/pinkstack/realestate/Domain.scala
+++ b/scraper/src/main/scala/com/pinkstack/realestate/scraper/Domain.scala
@@ -1,4 +1,4 @@
-package com.pinkstack.realestate
+package com.pinkstack.realestate.scraper
 
 import akka.http.scaladsl.model.{HttpRequest, Uri}
 import io.lemonlabs.uri.Url
diff --git a/src/main/scala/com/pinkstack/realestate/EstateParser.scala b/scraper/src/main/scala/com/pinkstack/realestate/scraper/EstateParser.scala
similarity index 98%
rename from src/main/scala/com/pinkstack/realestate/EstateParser.scala
rename to scraper/src/main/scala/com/pinkstack/realestate/scraper/EstateParser.scala
index e2bf868..24f4ce1 100644
--- a/src/main/scala/com/pinkstack/realestate/EstateParser.scala
+++ b/scraper/src/main/scala/com/pinkstack/realestate/scraper/EstateParser.scala
@@ -1,9 +1,9 @@
-package com.pinkstack.realestate
+package com.pinkstack.realestate.scraper
 
 import cats._
 import cats.data.{Validated, ValidatedNec}
 import cats.implicits._
-import com.pinkstack.realestate.Domain._
+import Domain._
 import io.lemonlabs.uri.Url
 import org.jsoup.nodes.Document
 
diff --git a/src/main/scala/com/pinkstack/realestate/NepClient.scala b/scraper/src/main/scala/com/pinkstack/realestate/scraper/NepClient.scala
similarity index 99%
rename from src/main/scala/com/pinkstack/realestate/NepClient.scala
rename to scraper/src/main/scala/com/pinkstack/realestate/scraper/NepClient.scala
index 9b7c8df..5ec82a3 100644
--- a/src/main/scala/com/pinkstack/realestate/NepClient.scala
+++ b/scraper/src/main/scala/com/pinkstack/realestate/scraper/NepClient.scala
@@ -1,4 +1,4 @@
-package com.pinkstack.realestate
+package com.pinkstack.realestate.scraper
 
 import cats._
 import cats.implicits._
diff --git a/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala b/scraper/src/main/scala/com/pinkstack/realestate/scraper/apps/Scraper.scala
similarity index 93%
rename from src/main/scala/com/pinkstack/realestate/apps/Scraper.scala
rename to scraper/src/main/scala/com/pinkstack/realestate/scraper/apps/Scraper.scala
index d728670..855657a 100644
--- a/src/main/scala/com/pinkstack/realestate/apps/Scraper.scala
+++ b/scraper/src/main/scala/com/pinkstack/realestate/scraper/apps/Scraper.scala
@@ -1,12 +1,12 @@
-package com.pinkstack.realestate.apps
+package com.pinkstack.realestate.scraper.apps
 
 import akka.actor.ActorSystem
 import akka.stream.scaladsl.Sink
 import cats.data._
 import cats.implicits._
 import com.monovore.decline._
-import com.pinkstack.realestate.Domain.Estate
-import com.pinkstack.realestate.{Configuration, NepClient}
+import com.pinkstack.realestate.scraper._
+import com.pinkstack.realestate.scraper.Domain._
 import io.circe.generic.auto._
 import io.circe.syntax._
 import monocle.macros.GenLens
diff --git a/src/test/resources/404-error-page.html b/scraper/src/test/resources/404-error-page.html
similarity index 100%
rename from src/test/resources/404-error-page.html
rename to scraper/src/test/resources/404-error-page.html
diff --git a/src/test/resources/dravograd-stanovanje.html b/scraper/src/test/resources/dravograd-stanovanje.html
similarity index 100%
rename from src/test/resources/dravograd-stanovanje.html
rename to scraper/src/test/resources/dravograd-stanovanje.html
diff --git a/src/test/resources/ljubljana-skladisce-no-price.html b/scraper/src/test/resources/ljubljana-skladisce-no-price.html
similarity index 100%
rename from src/test/resources/ljubljana-skladisce-no-price.html
rename to scraper/src/test/resources/ljubljana-skladisce-no-price.html
diff --git a/src/test/resources/novigrad-hisa.html b/scraper/src/test/resources/novigrad-hisa.html
similarity index 100%
rename from src/test/resources/novigrad-hisa.html
rename to scraper/src/test/resources/novigrad-hisa.html
diff --git a/src/test/resources/sela-posest.html b/scraper/src/test/resources/sela-posest.html
similarity index 100%
rename from src/test/resources/sela-posest.html
rename to scraper/src/test/resources/sela-posest.html
diff --git a/src/test/scala/NepEstateParserSpec.scala b/scraper/src/test/scala/NepEstateParserSpec.scala
similarity index 88%
rename from src/test/scala/NepEstateParserSpec.scala
rename to scraper/src/test/scala/NepEstateParserSpec.scala
index 363102c..2b772a4 100644
--- a/src/test/scala/NepEstateParserSpec.scala
+++ b/scraper/src/test/scala/NepEstateParserSpec.scala
@@ -4,8 +4,9 @@ import java.nio.file.Path
 import cats._
 import cats.data.Validated.{Invalid, Valid}
 import cats.implicits._
-import com.pinkstack.realestate.Domain.Estate
-import com.pinkstack.realestate.{NepEstateParser, NoTitleFound, NotActive}
+import com.pinkstack.realestate.scraper.Domain.Estate
+import com.pinkstack.realestate.scraper.NepEstateParser
+import com.pinkstack.realestate.scraper.NepEstateParser._
 import org.jsoup.Jsoup
 import org.jsoup.nodes.Document
 import org.scalatest.{EitherValues, PartialFunctionValues}
@@ -28,7 +29,7 @@ class NepEstateParserSpec extends AnyFlatSpec
 
   "An NepEstateParser" should "throw an error" in {
     read("404-error-page.html").map { doc =>
-      NepEstateParser.parse(doc) match {
+      parse(doc) match {
         case Invalid(e) => assert(true)
         case Valid(a) => fail("should fail not be valid")
       }
@@ -37,7 +38,7 @@ class NepEstateParserSpec extends AnyFlatSpec
 
   it should "regular listing" in {
     read("dravograd-stanovanje.html").map { doc =>
-      NepEstateParser.parse(doc).toEither match {
+      parse(doc).toEither match {
         case Right(Estate(title, refNumber, price, visibility, scrapedFromUrl, locationDetails, categories, location)) =>
           assert(title == "Prodaja, Stanovanje, 4-sobno: DRAVOGRAD, 146.2 m2")
           assert(categories("Posredovanje") == "Prodaja")
@@ -55,7 +56,7 @@ class NepEstateParserSpec extends AnyFlatSpec
 
     estates.map(e => (e._1, e._2, read(e._1))).foreach {
       case (name, expectedPrice, Some(document)) =>
-        NepEstateParser.parse(document).toEither match {
+        parse(document).toEither match {
           case Right(estate: Estate) =>
             assert(estate.price == expectedPrice)
           case Left(value) =>
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala b/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
deleted file mode 100644
index fe10bae..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/ClusteringV2.scala
+++ /dev/null
@@ -1,63 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor._
-import akka.cluster.Cluster
-import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, MemberRemoved, MemberUp, UnreachableMember}
-import com.typesafe.scalalogging.LazyLogging
-import com.typesafe.config.ConfigFactory
-
-class ClusterListenerActor extends Actor with ActorLogging {
-  val cluster = Cluster(context.system)
-
-  override def preStart(): Unit = {
-    cluster.subscribe(self,
-      initialStateMode = InitialStateAsEvents,
-      classOf[MemberEvent], classOf[UnreachableMember]
-    )
-  }
-
-  override def postStop(): Unit = {
-    cluster.unsubscribe(self)
-  }
-
-  override def receive: Receive = {
-    case MemberUp(member) =>
-      log.info("Member is up {}", member.address)
-    case UnreachableMember(member) =>
-      log.info("Member detected as unreachable {}", member.address)
-    case MemberRemoved(member, previousStatus) =>
-      log.info("Member removed {} after {}", member.address, previousStatus)
-    case event: MemberEvent =>
-      log.info("Got event {}", event)
-  }
-}
-
-object ClusteringV2 extends LazyLogging {
-  def main(args: Array[String]): Unit = {
-    logger.info("Clustering demo")
-
-    val clusteringConfig = ConfigFactory.load(
-      """
-        |akka {
-        | actor {
-        |   provider = "cluster"
-        | }
-        | remote.artery {
-        |    canonical {
-        |      hostname = "0.0.0.0"
-        |      port = 2551
-        |    }
-        | }
-        | cluster {
-        |    seed-nodes = [
-        |      "akka://ClusterSystem@127.0.0.1:2551",
-        |      "akka://ClusterSystem@127.0.0.1:2552"]
-        |
-        |   downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
-        | }
-        |}
-        |""".stripMargin)
-
-    println(clusteringConfig)
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
deleted file mode 100644
index dc2cb81..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/MainApp.scala
+++ /dev/null
@@ -1,32 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
-import akka.cluster.Cluster
-import akka.cluster.ClusterEvent.MemberEvent
-import com.typesafe.scalalogging.LazyLogging
-
-class ExampleClusterListener extends Actor with ActorLogging {
-  val cluster = Cluster(context.system)
-
-  override def receive: Receive = {
-    case event: MemberEvent =>
-      log.info(s"Received ${event}")
-  }
-}
-
-class PingActor extends Actor with ActorLogging {
-  override def receive: Receive = {
-    case s: String =>
-      log.info(s"Received $s in ${self}")
-  }
-}
-
-object MainApp extends App with LazyLogging {
-  implicit val system = ActorSystem("estateApp")
-  implicit val context = system.dispatcher
-
-  val pingA = system.actorOf(Props[PingActor])
-  pingA ! "Hello Oto"
-
-  println("This is the MainApp")
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
deleted file mode 100644
index 19050e9..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/MainApp2.scala
+++ /dev/null
@@ -1,46 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor.typed.scaladsl._
-import akka.actor.typed.ActorSystem
-import akka.actor.typed.Behavior
-import akka.cluster.typed.Cluster
-import akka.cluster.typed.Subscribe
-import com.typesafe.config.ConfigFactory
-import com.typesafe.scalalogging.LazyLogging
-
-object ClusterListener {
-
-  sealed trait Event
-
-  def apply(): Behavior[Event] = Behaviors.setup { ctx =>
-    Behaviors.receiveMessage {
-      case _ =>
-        println("ok")
-        Behaviors.same
-    }
-  }
-}
-
-object MainApp2 extends LazyLogging {
-
-  object RootBehavior {
-    def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] { context =>
-      context.spawn(ClusterListener(), "ClusterListener")
-      Behaviors.empty
-    }
-  }
-
-  def main(args: Array[String]): Unit = {
-    val ports: Seq[Int] = Seq(33200, 33201, 0)
-    ports.foreach(boot)
-  }
-
-  def boot(port: Int): Unit = {
-    val config = ConfigFactory.parseString(
-      s"""
-         |akka.remote.artery.canonical.host=0.0.0.0
-         |akka.remote.artery.canonical.port=$port""".stripMargin).withFallback(ConfigFactory.load)
-    ActorSystem[Nothing](RootBehavior(), s"ClusterSystem-$port", config)
-  }
-}
-
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
deleted file mode 100644
index 1593c38..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/MainApp3.scala
+++ /dev/null
@@ -1,58 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor.{ActorSystem, CoordinatedShutdown}
-import akka.discovery.{Discovery, Lookup}
-import com.typesafe.config.ConfigFactory
-import com.typesafe.scalalogging.LazyLogging
-import scala.concurrent.duration._
-
-object MainApp3 extends LazyLogging {
-  def main(args: Array[String]): Unit = {
-    val config = ConfigFactory.parseString(
-      s"""
-         |akka.discovery.method = aggregate
-         |akka.discovery.aggregate {
-         |  discovery-methods = ["akka-dns", "config"]
-         |}
-         |akka.discovery.config.services = {
-         |  seeder = {
-         |    endpoints = [
-         |      {
-         |        host = 0.0.0.0
-         |        port = 6000
-         |      }
-         |   ]
-         |  },
-         |  fetchers = {
-         |    endpoints = [
-         |      {
-         |        host = 0.0.0.0
-         |        port = 6001
-         |      },
-         |      {
-         |        host = 0.0.0.0
-         |        port = 6002
-         |      }
-         |    ]
-         |  }
-         |}
-         |akka.cluster.shutdown-after-unsuccessful-join-seed-nodes = 10s
-         |akka.coordinated-shutdown.exit-jvm = on
-         |""".stripMargin).withFallback(ConfigFactory.load)
-
-    implicit val system = ActorSystem("main3", config)
-    implicit val executionContext = system.dispatcher
-
-    val discovery = Discovery(system).discovery
-
-    discovery.lookup(Lookup("fetchers"), 3.seconds).map { a =>
-      println(a.getAddresses)
-    }
-
-    CoordinatedShutdown(system).addJvmShutdownHook {
-      logger.info("Terminating...")
-      // clean shutdown of the flow
-      logger.info("Terminated... Bye")
-    }
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala b/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala
deleted file mode 100644
index 7839c84..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/MainApp4.scala
+++ /dev/null
@@ -1,45 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor.typed._
-import akka.actor.typed.scaladsl._
-
-object Greeter {
-
-  final case class Greet(whom: String)
-
-  final case class Greeted(whom: String, from: ActorRef[Greet])
-
-  def apply(): Behavior[Greet] = Behaviors.receive { (context, message) =>
-    context.log.info(s"Hello ${message.whom}")
-
-    if (message.whom.contains("xxx")) {
-      Behaviors.stopped
-    } else Behaviors.same
-  }
-}
-
-object MainApp4 {
-
-  final case class Hello(name: String)
-
-  def apply(): Behavior[Hello] = Behaviors.setup { context =>
-    context.log.info("Setup")
-
-    val greeter = context.spawn(Greeter(), "greeter")
-
-    Behaviors.receiveMessage { message =>
-      context.log.info(s"Got $message")
-
-      greeter ! Greeter.Greet(s"hello ${message.name}")
-      Behaviors.same
-    }
-  }
-
-  def main(args: Array[String]): Unit = {
-    val system: ActorSystem[MainApp4.Hello] = ActorSystem(MainApp4(), "mainSystem")
-
-    system ! Hello("Oto")
-    system ! Hello("Martina")
-    system ! Hello("xxx")
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala b/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
deleted file mode 100644
index 660eb62..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering/StoreSystemApp.scala
+++ /dev/null
@@ -1,87 +0,0 @@
-package com.pinkstack.realestate.clustering
-
-import akka.actor.typed._
-import akka.actor.typed.scaladsl._
-import com.typesafe.scalalogging.LazyLogging
-
-object Domain {
-
-  sealed trait StoreMessage
-
-  case class AddItem(item: String) extends StoreMessage
-
-  case class RemoveItem(item: String) extends StoreMessage
-
-  case object ValidateCart extends StoreMessage
-
-}
-
-object StoreSystemApp extends LazyLogging {
-
-  import Domain._
-
-  def main(args: Array[String]): Unit = {
-    logger.info("StoreSystemApp booting,...")
-
-    val storeRootActor = ActorSystem(
-      Behaviors.receiveMessage[StoreMessage] { message =>
-        message match {
-          case AddItem(item) => println(s"Adding item to cart ${item}")
-          case RemoveItem(item) => println(s"Removing item ${item}")
-          case ValidateCart => println("cart is good")
-        }
-        Behaviors.same
-      }, "storeRootActor"
-    )
-
-    storeRootActor ! AddItem("cake")
-    storeRootActor ! AddItem("beer")
-
-
-    val storeRootActorMutable = ActorSystem(
-      Behaviors.setup[StoreMessage] { ctx =>
-        // Local state
-        var items: Set[String] = Set()
-
-        Behaviors.receiveMessage[StoreMessage] { message =>
-          message match {
-            case AddItem(item) =>
-              items += item
-              println(s"Item ${item} added.")
-            case RemoveItem(item) =>
-              items -= item
-              println(s"Removed item ${item}")
-            case ValidateCart => println("cart is good")
-          }
-          Behaviors.same
-        }
-      }
-      , "storeRootActorMutalbe"
-    )
-
-    def shoppingBehaviour(items: Set[String]): Behavior[StoreMessage] =
-      Behaviors.receiveMessage[StoreMessage] {
-        case AddItem(item) =>
-          println("adding")
-          shoppingBehaviour(items + item)
-        case RemoveItem(item) =>
-          println("removing")
-          shoppingBehaviour(items - item)
-        case ValidateCart =>
-          Behaviors.same
-      }
-
-    storeRootActorMutable ! AddItem("coke")
-    storeRootActorMutable ! AddItem("coke")
-    storeRootActorMutable ! AddItem("beer")
-
-    val rootOnlineActor = ActorSystem(
-      Behaviors.setup[StoreMessage] { ctx =>
-        ctx.spawn(shoppingBehaviour(Set()), "otoCart")
-        Behaviors.empty
-      }, "onlineStore"
-    )
-
-    rootOnlineActor ! AddItem("xxx")
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
deleted file mode 100644
index 76cfc6a..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterListener.scala
+++ /dev/null
@@ -1,21 +0,0 @@
-package com.pinkstack.realestate.clustering3
-
-import akka.actor.{Actor, ActorLogging}
-import akka.cluster.Cluster
-import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, UnreachableMember}
-import com.pinkstack.realestate.clustering3.Domain.NodeID
-
-class ClusterListener(nodeID: NodeID, cluster: Cluster) extends Actor with ActorLogging {
-  override def preStart(): Unit =
-    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
-      classOf[MemberEvent], classOf[UnreachableMember])
-
-  override def postStop(): Unit = cluster.unsubscribe(self)
-
-  override def receive: Receive = {
-    case x: Any =>
-      println("~~~>")
-      println(x)
-      println("~~~>")
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
deleted file mode 100644
index 107ed38..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering3/ClusterManager.scala
+++ /dev/null
@@ -1,36 +0,0 @@
-package com.pinkstack.realestate.clustering3
-
-import akka.actor.{Actor, ActorLogging, ActorRef, Props}
-import akka.cluster.{Cluster, Member}
-import com.pinkstack.realestate.clustering3.Domain.NodeID
-
-class ClusterManager(nodeID: NodeID) extends Actor with ActorLogging {
-  val cluster: Cluster = Cluster(context.system)
-  val listener: ActorRef = context.actorOf(Props(classOf[ClusterListener], nodeID, cluster), "clusterListener")
-
-  import ClusterManager._
-
-  override def receive: Receive = {
-    case GetMembers =>
-      sender ! MembersCollected {
-        cluster.state.members.toList.map { member: Member =>
-          Map[String, String](
-            "address" -> member.address.toString,
-            "status" -> member.status.toString
-          )
-        }
-      }
-  }
-}
-
-object ClusterManager {
-
-  sealed trait ClusterMessage
-
-  sealed trait ClusterEvent
-
-  case object GetMembers extends ClusterMessage
-
-  case class MembersCollected(members: List[Map[String, String]]) extends ClusterEvent
-
-}
\ No newline at end of file
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala b/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
deleted file mode 100644
index 3f52d8a..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering3/Domain.scala
+++ /dev/null
@@ -1,12 +0,0 @@
-package com.pinkstack.realestate.clustering3
-
-object Domain {
-  type NodeID = String
-
-  sealed trait Command
-
-  case class FetchEstate(url: String) extends Command
-
-  case object ScrapeAll extends Command
-
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala b/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
deleted file mode 100644
index b94cef4..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering3/Node.scala
+++ /dev/null
@@ -1,33 +0,0 @@
-package com.pinkstack.realestate.clustering3
-
-import akka.actor.{Actor, ActorLogging, ActorRef, Props}
-import com.pinkstack.realestate.clustering3.ClusterManager.GetMembers
-import com.pinkstack.realestate.clustering3.Domain.{FetchEstate, NodeID, ScrapeAll}
-import com.pinkstack.realestate.{Configuration, NepClient}
-import io.lemonlabs.uri.Url
-
-import scala.concurrent.ExecutionContext
-import scala.concurrent.duration._
-import akka.pattern.{ask, pipe}
-
-class Node(nodeID: NodeID) extends Actor with ActorLogging {
-  val configuration: Configuration = Configuration.loadOrThrow
-  val nepClient: NepClient = NepClient(4.seconds)(this.context.system, this.context.system.dispatcher, configuration)
-  implicit val executionContext: ExecutionContext = this.context.dispatcher
-
-  val clusterManager: ActorRef = context.actorOf(Props(classOf[ClusterManager], nodeID), "clusterManager")
-
-  override def preStart(): Unit = log.info("preStart of {} in {}", nodeID, self)
-
-  override def receive: Receive = {
-    case FetchEstate(url) =>
-      log.info("Fetching {}", url)
-      nepClient.fetchEstatePageViaUrl(Url.parse(url)) pipeTo sender
-    case command@GetMembers =>
-      ((clusterManager ? command) (3.seconds)) pipeTo sender
-    case ScrapeAll =>
-
-    case x: Any =>
-      log.info("Got {} at {}", x, self)
-  }
-}
diff --git a/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala b/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
deleted file mode 100644
index 7a3f47d..0000000
--- a/src/main/scala/com/pinkstack/realestate/clustering3/ServerApp.scala
+++ /dev/null
@@ -1,88 +0,0 @@
-package com.pinkstack.realestate.clustering3
-
-import akka.actor._
-import com.typesafe.config.{Config, ConfigFactory}
-
-import scala.concurrent.{Await, ExecutionContext, Future}
-import scala.concurrent.duration._
-import akka.http.scaladsl._
-import akka.http.scaladsl.model.{HttpRequest, StatusCodes, Uri}
-import akka.http.scaladsl.server.Route
-import akka.http.scaladsl.server.Directives._
-import com.pinkstack.realestate.clustering3.ClusterManager.{GetMembers, MembersCollected}
-import com.typesafe.scalalogging.LazyLogging
-import akka.pattern.ask
-import akka.pattern.pipe
-import akka.pattern.PipeToSupport
-import akka.util.Timeout
-import com.pinkstack.realestate.Domain.{Estate, EstateRequest}
-import com.pinkstack.realestate.{Configuration, NepClient}
-import io.lemonlabs.uri.Url
-
-import Domain._
-
-
-trait ServerRoutes {
-  this: LazyLogging =>
-
-  import de.heikoseeberger.akkahttpcirce.FailFastCirceSupport._
-  import io.circe.generic.auto._
-
-  implicit val system: ActorSystem
-  implicit val dispatcher: ExecutionContext
-
-  def stopSystem: Future[String] = for {
-    _ <- system.terminate()
-    _ = logger.info("System terminated.")
-  } yield "Terminated"
-
-  val fetchEstate: (ActorRef, String) => Future[Option[Estate]] = (nodeActor, url) =>
-    (nodeActor ? FetchEstate(url)) (3.seconds).mapTo[Option[Estate]]
-
-  val getMembers: ActorRef => Future[MembersCollected] = { nodeActorRef =>
-    (nodeActorRef ? GetMembers) (3.seconds).mapTo[MembersCollected]
-  }
-  val scrapeAll: ActorRef => Future[Map[String, String]] = { nodeActorRef =>
-    nodeActorRef ! ScrapeAll
-    Future.successful(Map[String, String]("scraping" -> "started,..."))
-  }
-
-  def basicRoutes(nodeActorRef: ActorRef): Route = {
-    concat(
-      pathSingleSlash(get(complete(StatusCodes.OK, "ok"))),
-      path("stop")(get(complete(stopSystem))),
-      path("fetchEstate") {
-        parameters(Symbol("url").as[String]) { (url: String) =>
-          onSuccess(fetchEstate(nodeActorRef, url)) {
-            case Some(estate) => complete(StatusCodes.OK, estate)
-            case None => complete(StatusCodes.BadRequest, Map("error" -> "Sorry, fetching failed."))
-          }
-        }
-      },
-      path("getMembers")(get {
-        onSuccess(getMembers(nodeActorRef)) { result: MembersCollected =>
-          complete(Map("members" -> result.members))
-        }
-      }),
-      path("scrape")(get(complete(scrapeAll(nodeActorRef)))
-      )
-    )
-  }
-}
-
-object ServerApp extends App with LazyLogging with ServerRoutes {
-  val config: Config = ConfigFactory.load("clusteringv3.conf")
-  implicit val system: ActorSystem = ActorSystem("mySystem", config)
-  implicit val dispatcher: ExecutionContext = system.dispatcher
-
-  val address: String = config.getString("http.ip")
-  val port: Int = config.getInt("http.port")
-
-  val node: ActorRef = system.actorOf(Props(classOf[Node], "x"), "node")
-
-  Http().bindAndHandle(basicRoutes(node), address, port)
-  logger.info(s"🚀 Node is listening on http://{}:{}", address, port)
-  logger.info(s"🛑 Stop it at http://{}:{}/stop", address, port)
-
-  Await.result(system.whenTerminated, Duration.Inf)
-}
diff --git a/worksheets/coordinates.sc b/worksheets/coordinates.sc
deleted file mode 100644
index 0bc50db..0000000
--- a/worksheets/coordinates.sc
+++ /dev/null
@@ -1,11 +0,0 @@
-val coords = "(45.9818666,14.2267596)"
-
-val c =
-  """(-?\d+(\.\d+)?),\s*(-?\d+(\.\d+)?)""".r.findFirstIn(coords)
-    .map(_.split(",", 2).map(_.toDouble).toList)
-
-c match {
-  case Some(v) => v
-  case _ => "nopu"
-}
-
diff --git a/worksheets/parsing.sc b/worksheets/parsing.sc
deleted file mode 100644
index b8acb45..0000000
--- a/worksheets/parsing.sc
+++ /dev/null
@@ -1,16 +0,0 @@
-import org.jsoup.Jsoup
-
-val documents: Seq[String] =
-  "https://www.nepremicnine.net/oglasi-prodaja/medulin-stanovanje_6347567/" ::
-    "https://www.nepremicnine.net/oglasi-prodaja/sezana-garaza_6347637/" ::
-    "https://www.nepremicnine.net/oglasi-prodaja/sestdobe-posest_6347670/" ::
-    Nil
-
-
-
-// Jsoup.connect(documents.head).get()
-
-"this".split(",") match {
-  case Array(xs) => "x"
-  case _ => "y"
-}
\ No newline at end of file
diff --git a/worksheets/ranges.sc b/worksheets/ranges.sc
deleted file mode 100644
index e7f4b12..0000000
--- a/worksheets/ranges.sc
+++ /dev/null
@@ -1 +0,0 @@
-Range(1, 10).toList
\ No newline at end of file

From 1e556b84f9a43e5c01a37c5bffc2d96edf152df3 Mon Sep 17 00:00:00 2001
From: Oto Brglez <otobrglez@gmail.com>
Date: Tue, 14 Jul 2020 00:23:51 +0200
Subject: [PATCH 4/4] Update README.md

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 44f28e4..9cb8551 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@ 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].