Influxdb connector

.travis.yml

@@ -73,6 +73,9 @@ jobs:
  - PRE_CMD="docker-compose up -d hbase"
  - env:
  - DIR=hdfs
+ - env:
+ - DIR=influxdb
+ - PRE_CMD="docker-compose up -d influxdb"
  - env:
  - DIR=ironmq
  - PRE_CMD="docker-compose up -d ironauth ironmq"

build.sbt

@@ -18,6 +18,7 @@ lazy val modules: Seq[ProjectReference] = Seq(
  googleFcm,
  hbase,
  hdfs,
+ influxdb,
  ironmq,
  jms,
  jsonStreaming,
@@ -178,6 +179,8 @@ lazy val hbase = alpakkaProject("hbase", "hbase", Dependencies.HBase, fork in Te
 
 lazy val hdfs = alpakkaProject("hdfs", "hdfs", Dependencies.Hdfs, parallelExecution in Test := false)
 
+lazy val influxdb = alpakkaProject("influxdb", "influxdb", Dependencies.InfluxDB).disablePlugins(MimaPlugin)
+
 lazy val ironmq = alpakkaProject("ironmq",
  "ironmq",
  Dependencies.IronMq,

docker-compose.yml

@@ -202,6 +202,10 @@ services:
  image: mongo
  ports:
  - "27017:27017"
+ influxdb:
+ image: influxdb
+ ports:
+ - "8086:8086"
  mqtt:
  image: toke/mosquitto
  ports:

influxdb/src/main/java/org/influxdb/impl/InfluxDBResultMapperHelper.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package org.influxdb.impl;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.influxdb.InfluxDBMapperException;
+import org.influxdb.dto.QueryResult;
+
+public class InfluxDBResultMapperHelper {
+
+ private final InfluxDBResultMapper influxDBResultMapper = new InfluxDBResultMapper();
+
+ public void cacheClassFields(final Class<?> clazz) {
+ influxDBResultMapper.cacheMeasurementClass(clazz);
+ }
+
+ public <T> List<T> parseSeriesAs(
+ final Class<T> clazz, final QueryResult.Series series, final TimeUnit precision) {
+ influxDBResultMapper.cacheMeasurementClass(clazz);
+ return series
+ .getValues()
+ .stream()
+ .map(v -> parseRowAs(clazz, series.getColumns(), v, precision))
+ .collect(Collectors.toList());
+ }
+
+ public <T> T parseRowAs(
+ final Class<T> clazz, List<String> columns, final List<Object> values, TimeUnit precision) {
+
+ try {
+ ConcurrentMap<String, Field> fieldMap = influxDBResultMapper.getColNameAndFieldMap(clazz);
+
+ T object = null;
+
+ for (int i = 0; i < columns.size(); i++) {
+ Field correspondingField = fieldMap.get(columns.get(i));
+
+ if (correspondingField != null) {
+ if (object == null) {
+ object = clazz.newInstance();
+ }
+
+ influxDBResultMapper.setFieldValue(object, correspondingField, values.get(i), precision);
+ }
+ }
+
+ return object;
+ } catch (InstantiationException | IllegalAccessException e) {
+ throw new InfluxDBMapperException(e);
+ }
+ }
+}

influxdb/src/main/resources/reference.conf

@@ -0,0 +1,9 @@
+akka.stream.alpakka.influxdb{
+
+ #batch write size
+ batchSize = 10
+
+ # The InfluxDB write precision
+ precision = MILLISECONDS
+
+}

influxdb/src/main/scala/akka/stream/alpakka/influxdb/InfluxDBSettings.scala

@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.influxdb
+
+import java.util.concurrent.TimeUnit
+
+import scala.concurrent.duration.TimeUnit
+
+object InfluxDBSettings {
+ val Default = new InfluxDBSettings(batchSize = 10, TimeUnit.MILLISECONDS)
+
+ def apply(): InfluxDBSettings = Default
+
+}
+
+final class InfluxDBSettings private (
+ val batchSize: Int,
+ val precision: TimeUnit
+) {
+
+ def withBatchSize(value: Int): InfluxDBSettings = copy(batchSize = value)
+
+ private def copy(
+ batchSize: Int = batchSize,
+ precision: TimeUnit = precision
+ ): InfluxDBSettings = new InfluxDBSettings(
+ batchSize = batchSize,
+ precision = precision
+ )
+
+}

influxdb/src/main/scala/akka/stream/alpakka/influxdb/InfluxDBWriteMessage.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.influxdb
+
+import akka.NotUsed
+
+object InfluxDBWriteMessage {
+ // Apply method to use when not using passThrough
+ def apply[T](point: T): InfluxDBWriteMessage[T, NotUsed] =
+ InfluxDBWriteMessage(point, NotUsed)
+
+ // Java-api - without passThrough
+ def create[T](point: T): InfluxDBWriteMessage[T, NotUsed] =
+ InfluxDBWriteMessage(point, NotUsed)
+
+ // Java-api - with passThrough
+ def create[T, C](point: T, passThrough: C) =
+ InfluxDBWriteMessage(point, passThrough)
+}
+
+final case class InfluxDBWriteMessage[T, C](point: T, passThrough: C)
+
+final case class InfluxDBWriteResult[T, C](writeMessage: InfluxDBWriteMessage[T, C], error: Option[String])

influxdb/src/main/scala/akka/stream/alpakka/influxdb/impl/InfluxDBFlowStage.scala

@@ -0,0 +1,85 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.influxdb.impl
+
+import akka.stream.{Attributes, FlowShape, Inlet, Outlet}
+import akka.stream.alpakka.influxdb.{InfluxDBWriteMessage, InfluxDBWriteResult}
+import akka.stream.stage.{GraphStage, GraphStageLogic, InHandler, OutHandler}
+import org.influxdb.InfluxDB
+import org.influxdb.impl.InfluxDBMapper
+
+import scala.collection.immutable
+import org.influxdb.BatchOptions
+import org.influxdb.dto.Point
+
+/**
+ * INTERNAL API
+ */
+private[influxdb] class InfluxDBFlowStage[T, C](
+ clazz: Option[Class[T]],
+ influxDB: InfluxDB
+) extends GraphStage[FlowShape[immutable.Seq[InfluxDBWriteMessage[T, C]], immutable.Seq[InfluxDBWriteResult[T, C]]]] {
+ private val in = Inlet[immutable.Seq[InfluxDBWriteMessage[T, C]]]("in")
+ private val out = Outlet[immutable.Seq[InfluxDBWriteResult[T, C]]]("out")
+
+ override val shape = FlowShape(in, out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ clazz match {
+ case Some(c) => new InfluxDBMapperRecordLogic
+ case None => new InfluxDBRecordLogic
+ }
+
+ sealed abstract class InfluxDBLogic extends GraphStageLogic(shape) with InHandler with OutHandler {
+
+ protected def write(messages: immutable.Seq[InfluxDBWriteMessage[T, C]]): Unit
+
+ setHandlers(in, out, this)
+
+ override def onPull(): Unit = if (!isClosed(in) && !hasBeenPulled(in)) pull(in)
+
+ override def onPush(): Unit = {
+ val messages = grab(in)
+ if (messages.nonEmpty) {
+
+ influxDB.enableBatch(BatchOptions.DEFAULTS)
+ write(messages)
+ val writtenMessages = messages.map(m => new InfluxDBWriteResult(m, Option.empty))
+ influxDB.close()
+ push(out, writtenMessages)
+ }
+
+ tryPull(in)
+ }
+ }
+
+ final class InfluxDBRecordLogic extends InfluxDBLogic {
+
+ override protected def write(messages: immutable.Seq[InfluxDBWriteMessage[T, C]]): Unit =
+ messages.foreach {
+ case InfluxDBWriteMessage(point: Point, _) => {
+ influxDB.write(point)
+ }
+ case InfluxDBWriteMessage(others: AnyRef, _) =>
+ failStage(new RuntimeException(s"unexpected type Point or annotated with Measurement required"))
+ }
+
+ }
+
+ final class InfluxDBMapperRecordLogic extends InfluxDBLogic {
+
+ protected var influxDBMapper: InfluxDBMapper = _
+
+ override def preStart(): Unit =
+ influxDBMapper = new InfluxDBMapper(influxDB)
+
+ override protected def write(messages: immutable.Seq[InfluxDBWriteMessage[T, C]]): Unit =
+ messages.foreach {
+ case InfluxDBWriteMessage(typeMetric: Any, _) =>
+ influxDBMapper.save(typeMetric)
+ }
+ }
+
+}

influxdb/src/main/scala/akka/stream/alpakka/influxdb/impl/InfluxDBSourceStage.scala

@@ -0,0 +1,128 @@
+/*
+ * Copyright (C) 2016-2019 Lightbend Inc. <http://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.influxdb.impl
+
+import java.util.concurrent.TimeUnit
+
+import akka.annotation.InternalApi
+import akka.stream.alpakka.influxdb.InfluxDBSettings
+import akka.stream.{Attributes, Outlet, SourceShape}
+import akka.stream.stage.{GraphStage, GraphStageLogic, OutHandler}
+import org.influxdb.InfluxDB
+import org.influxdb.dto.{Query, QueryResult}
+import org.influxdb.impl.InfluxDBResultMapperHelper
+
+import scala.collection.JavaConverters._
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[influxdb] final class InfluxDBSourceStage[T](clazz: Class[T],
+ settings: InfluxDBSettings,
+ influxDB: InfluxDB,
+ query: Query)
+ extends GraphStage[SourceShape[T]] {
+
+ val out: Outlet[T] = Outlet("InfluxDB.out")
+ override val shape = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new InfluxDBSourceLogic[T](clazz, settings, influxDB, query, out, shape)
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[influxdb] final class InfluxDBSourceLogic[T](clazz: Class[T],
+ settings: InfluxDBSettings,
+ influxDB: InfluxDB,
+ query: Query,
+ outlet: Outlet[T],
+ shape: SourceShape[T])
+ extends GraphStageLogic(shape)
+ with OutHandler {
+
+ setHandler(outlet, this)
+
+ var dataRetrieved: Option[QueryResult] = Option.empty
+ var resultMapperHelper: InfluxDBResultMapperHelper = _
+
+ override def preStart(): Unit = {
+ resultMapperHelper = new InfluxDBResultMapperHelper
+ resultMapperHelper.cacheClassFields(clazz)
+
+ val queryResult = influxDB.query(query)
+ if (!queryResult.hasError) {
+ dataRetrieved = Option(queryResult)
+ }
+ }
+
+ override def onPull(): Unit =
+ if (dataRetrieved.isEmpty)
+ completeStage()
+ else {
+ val queryResult = dataRetrieved.get
+ for {
+ result <- queryResult.getResults.asScala
+ series <- result.getSeries.asScala
+ }(
+ emitMultiple(outlet,
+ resultMapperHelper.parseSeriesAs(clazz, series, settings.precision).asScala.toIterator)
+ )
+
+ dataRetrieved = Option.empty
+ }
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[influxdb] final class InfluxDBRawSourceStage(query: Query, influxDB: InfluxDB)
+ extends GraphStage[SourceShape[QueryResult]] {
+
+ val out: Outlet[QueryResult] = Outlet("InfluxDB.out")
+ override val shape = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new InfluxDBSourceRawLogic(query, influxDB, out, shape)
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[influxdb] final class InfluxDBSourceRawLogic(query: Query,
+ influxDB: InfluxDB,
+ outlet: Outlet[QueryResult],
+ shape: SourceShape[QueryResult])
+ extends GraphStageLogic(shape)
+ with OutHandler {
+
+ setHandler(outlet, this)
+
+ var dataRetrieved: Option[QueryResult] = Option.empty
+
+ override def preStart(): Unit = {
+ val queryResult = influxDB.query(query)
+ if (!queryResult.hasError) {
+ dataRetrieved = Option(queryResult)
+ }
+ }
+
+ override def onPull(): Unit =
+ if (dataRetrieved.isEmpty) {
+ completeStage()
+ } else {
+ emit(outlet, dataRetrieved.get)
+ dataRetrieved = Option.empty
+ }
+
+}