ftrossbach

{  
   "hostInfo":{  
      "host":"1.2.3.4",
      "port":42
   },
   "stateStoreNames":[  
      "totalVisitCount",
      "hourlyVisitCount",
      "sessionVisitCount"
   ],

ftrossbach

KStreamBuilder builder = new KStreamBuilder();
KStream<String, Long> visitsStream = builder.stream(Serdes.String(), Serdes.Long(), "visitsTopic");
KGroupedStream<String, Long> groupedStream = visitsStream.groupByKey();
KTable<String, Long> totalCount = groupedStream.count("totalVisitCount");
KTable<Windowed<String>, Long> windowedCount = groupedStream.count(TimeWindows.of(60 * 60 * 1000), "hourlyVisitCount");
groupedStream.count(SessionWindows.with(60 * 1000), "sessionVisitCount");

ftrossbach

package de.ftrossbach.meetup

import java.util.concurrent.TimeUnit

import akka.actor.ActorSystem
import akka.{Done, NotUsed}
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl.model._