jeroenr

@Document
data class UserAccount(
	@Id
	val _id: ObjectId = ObjectId(),
	var name: String,
	var createdAt: Instant = Instant.now(),
	var updatedAt: Instant = Instant.now()
) {
	var auditTrail: List<String> = listOf()
        fun updateName(name: String) {

jeroenr

enum class Currency { USD, EUR }

data class Money(
        val amount: BigDecimal,
        val currency: Currency,
) {
    val largerThanZero = amount > BigDecimal.ZERO
    fun add(o: Money): Money {
        if(currency != o.currency) throw IllegalArgumentException()
        return Money(amount.add(o.amount), currency)

jeroenr

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer

object Boot extends App with Service with CorsSupport {
  override implicit val system = ActorSystem()
  override implicit val executor = system.dispatcher
  override implicit val materializer = ActorMaterializer()

  Http().bindAndHandle(corsHandler(routes), "0.0.0.0", 1337)

jeroenr

apiVersion: v1
kind: Namespace
metadata:
  name: monitoring
  labels:
    name: monitoring
---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:

jeroenr

package com.github.jeroenr.rain.radar

import akka.event.Logging
import cloudflow.akkastream._
import cloudflow.akkastream.scaladsl._
import cloudflow.streamlets._
import cloudflow.streamlets.avro._
import org.apache.avro.specific.SpecificRecordBase

import scala.reflect.ClassTag

jeroenr

package com.github.jeroenr.rain.radar

import cloudflow.akkastream._
import cloudflow.akkastream.util.scaladsl._
import cloudflow.streamlets._
import cloudflow.streamlets.avro._

class RainClutterPartitioner extends AkkaStreamlet {
  val in = AvroInlet[PrecipitationData]("in")
  val clutter = AvroOutlet[Clutter]("clutter").withPartitioner(RoundRobinPartitioner)

jeroenr

package com.github.jeroenr.rain.radar

import java.time.Instant

import spray.json._

trait InstantJsonSupport extends DefaultJsonProtocol {
  implicit object InstantFormat extends JsonFormat[Instant] {
    def write(instant: Instant) = JsNumber(instant.toEpochMilli)

jeroenr

import sbt._
import sbt.Keys._

lazy val rainRadar =  (project in file("."))
    .enablePlugins(CloudflowAkkaStreamsApplicationPlugin)
    .settings(
      libraryDependencies ++= Seq(
        "com.typesafe.akka"      %% "akka-http-spray-json"      % "10.1.10",
        "ch.qos.logback"         %  "logback-classic"           % "1.2.3"
      ),

jeroenr

blueprint {
  streamlets {
    http-ingress = com.github.jeroenr.rain.radar.PrecipitationDataHttpIngress
    partitioner = com.github.jeroenr.rain.radar.RainClutterPartitioner
    rain-logger = com.github.jeroenr.rain.radar.RainLogger
    clutter-logger = com.github.jeroenr.rain.radar.ClutterLogger
  }

  connections {
    http-ingress.out = [partitioner.in]

jeroenr

package com.github.jeroenr.rain.radar

import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import cloudflow.akkastream._
import cloudflow.akkastream.util.scaladsl._
import cloudflow.streamlets._
import cloudflow.streamlets.avro._
import com.github.jeroenr.rain.radar.PrecipitationDataJsonSupport._

class PrecipitationDataHttpIngress extends AkkaServerStreamlet {