feat: add Kotlin WriteApi (#222)

Author: bednar

CHANGELOG.md

@@ -14,7 +14,14 @@ You have to replace your dependency from: `influxdb-client-scala` to:
 1. [#218](https://github.com/influxdata/influxdb-client-java/pull/218): Supports enum types in mapping into POJO
 1. [#220](https://github.com/influxdata/influxdb-client-java/pull/220): Create client supporting OSGi environments
 1. [#221](https://github.com/influxdata/influxdb-client-java/pull/221): Add feature definition and documentation for Apache Karaf support
+1. [#222](https://github.com/influxdata/influxdb-client-java/pull/221): Add `Kotlin` WriteApi
 
+### Dependencies
+1. [#222](https://github.com/influxdata/influxdb-client-csharp/pull/222): Update dependencies:
+   - Kotlin to 1.4.32
+1. [#222](https://github.com/influxdata/influxdb-client-csharp/pull/222): Update plugins:
+   - dokka-maven-plugin to 1.4.30
+   
 ## 2.1.0 [2021-04-01]
 
 ### Bug Fixes

README.md

@@ -48,7 +48,7 @@ The Java, Reactive, OSGi, Kotlin and Scala clients are implemented for the Influ
 | --- | --- | --- |                                      --- |
 | **[java](./client)** | The reference Java client that allows query, write and InfluxDB 2.0 management. |  [javadoc](https://influxdata.github.io/influxdb-client-java/influxdb-client-java/apidocs/index.html), [readme](./client#influxdb-client-java/)| 2.0 |
 | **[reactive](./client-reactive)**  | The reference RxJava client for the InfluxDB 2.0 that allows query and write in a reactive way.| [javadoc](https://influxdata.github.io/influxdb-client-java/influxdb-client-java/apidocs/index.html), [readme](./client#influxdb-client-java/) |2.0 |
-| **[kotlin](./client-kotlin)** | The reference Kotlin client that allows query and write for the InfluxDB 2.0 by [Kotlin Channel coroutines](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-channel/index.html). | [KDoc](https://influxdata.github.io/influxdb-client-java/influxdb-client-kotlin/dokka/influxdb-client-kotlin/com.influxdb.client.kotlin/index.html), [readme](./client-kotlin#influxdb-client-kotlin/) | 2.0|
+| **[kotlin](./client-kotlin)** | The reference Kotlin client that allows query and write for the InfluxDB 2.0 by Kotlin [Channel](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-channel/index.html) and [Flow](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/index.html) coroutines. | [KDoc](https://influxdata.github.io/influxdb-client-java/influxdb-client-kotlin/dokka/influxdb-client-kotlin/com.influxdb.client.kotlin/index.html), [readme](./client-kotlin#influxdb-client-kotlin/) | 2.0|
 | **[scala](./client-scala)** | The reference Scala client that allows query and write for the InfluxDB 2.0 by [Akka Streams](https://doc.akka.io/docs/akka/2.6/stream/). | [Scaladoc](https://influxdata.github.io/influxdb-client-java/client-scala/cross/influxdb-client-scala_2.13/scaladocs/com/influxdb/client/scala/index.html), [readme](./client-scala#influxdb-client-scala/) | 2.0 |
 | **[osgi](./client-osgi)** | The reference OSGi (R6) client embedding Java and reactive clients and providing standard features (declarative services, configuration, event processing) for the InfluxDB 2.0. | [javadoc](https://influxdata.github.io/influxdb-client-java/influxdb-client-osgi/apidocs/index.html), [readme](./client-osgi) | 2.0 |
 | **[karaf](./karaf)** | The Apache Karaf feature definition for the InfluxDB 2.0. | [readme](./karaf) | 2.0 |

client-kotlin/README.md

@@ -7,6 +7,7 @@ The reference Kotlin client that allows query and write for the InfluxDB 2.0 by
 ## Features
 
 - [Querying data using Flux language](#queries)
+- [Writing data](#writes)
 - [Advanced Usage](#advanced-usage)
 
 ## Queries
@@ -78,6 +79,89 @@ fun main(args: Array<String>) = runBlocking {
 }
 ```
 
+## Writes
+
+The [WriteKotlinApi](https://influxdata.github.io/influxdb-client-java/influxdb-client-kotlin/dokka/influxdb-client-kotlin/com.influxdb.client.kotlin/-write-kotlin-api/index.html) supports ingest data by:
+- `DataPoint`
+- `LineProtocol`
+- `Data class`
+- List of above items
+
+The following example shows how to use various type of data:
+
+```kotlin
+package example
+
+import com.influxdb.annotations.Column
+import com.influxdb.annotations.Measurement
+import com.influxdb.client.domain.WritePrecision
+import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
+import com.influxdb.client.write.Point
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.runBlocking
+import java.time.Instant
+
+fun main() = runBlocking {
+
+    val org = "my-org"
+    val bucket = "my-bucket"
+
+    //
+    // Initialize client
+    //
+    val client = InfluxDBClientKotlinFactory
+        .create("http://localhost:8086", "my-token".toCharArray(), org, bucket)
+
+    val writeApi = client.getWriteKotlinApi()
+
+    //
+    // Write by Data Point
+    //
+    val point = Point.measurement("temperature")
+        .addTag("location", "west")
+        .addField("value", 55.0)
+        .time(Instant.now().toEpochMilli(), WritePrecision.MS)
+
+    writeApi.writePoint(point)
+
+    //
+    // Write by LineProtocol
+    //
+    writeApi.writeRecord("temperature,location=north value=60.0", WritePrecision.NS)
+
+    //
+    // Write by DataClass
+    //
+    val temperature = Temperature("south", 62.0, Instant.now())
+
+    writeApi.writeMeasurement(temperature, WritePrecision.NS)
+
+    //
+    // Query results
+    //
+    val fluxQuery =
+        """from(bucket: "$bucket") |> range(start: 0) |> filter(fn: (r) => (r["_measurement"] == "temperature"))"""
+
+    client
+        .getQueryKotlinApi()
+        .query(fluxQuery)
+        .consumeAsFlow()
+        .collect { println("Measurement: ${it.measurement}, value: ${it.value}") }
+
+    client.close()
+}
+
+@Measurement(name = "temperature")
+data class Temperature(
+    @Column(tag = true) val location: String,
+    @Column val value: Double,
+    @Column(timestamp = true) val time: Instant
+)
+
+```
+* sources - [KotlinWriteApi.kt](../examples/src/main/java/example/KotlinWriteApi.kt)
+
 ## Advanced Usage
 
 ### Client configuration file

client-kotlin/pom.xml

@@ -35,7 +35,8 @@
 
     <name>The Kotlin InfluxDB 2.0 Client</name>
     <description>
-        The reference Kotlin client that allows query and write for the InfluxDB 2.0 by Kotlin Channel coroutines.
+        The reference Kotlin client that allows query and write for the InfluxDB 2.0
+        by Kotlin Channel and Flow coroutines.
     </description>
 
     <url>https://github.com/influxdata/influxdb-client-java/tree/master/client-kotlin</url>
@@ -107,7 +108,7 @@
             <plugin>
                 <groupId>org.jetbrains.dokka</groupId>
                 <artifactId>dokka-maven-plugin</artifactId>
-                <version>0.10.1</version>
+                <version>1.4.30</version>
                 <executions>
                     <execution>
                         <id>dokka-pre-site</id>

client-kotlin/src/main/kotlin/com/influxdb/client/kotlin/InfluxDBClientKotlin.kt

@@ -39,6 +39,13 @@ interface InfluxDBClientKotlin : Closeable {
      */
     fun getQueryKotlinApi() : QueryKotlinApi
 
+    /**
+     * Get the Write client.
+     *
+     * @return the new client instance for the Write API
+     */
+    fun getWriteKotlinApi() : WriteKotlinApi
+
     /**
      * Get the health of an instance.
      *

client-kotlin/src/main/kotlin/com/influxdb/client/kotlin/QueryKotlinApi.kt

@@ -35,47 +35,52 @@ import kotlinx.coroutines.channels.Channel
 interface QueryKotlinApi {
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream [FluxRecord]s to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream
+     * [com.influxdb.query.FluxRecord]s to [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
-     * @return the stream of [FluxRecord]s
+     * @return the stream of [com.influxdb.query.FluxRecord]s
      */
     fun query(query: String): Channel<FluxRecord>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream [FluxRecord]s to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream
+     * [com.influxdb.query.FluxRecord]s to [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization
-     * @return the stream of [FluxRecord]s
+     * @return the stream of [com.influxdb.query.FluxRecord]s
      */
     fun query(query: String, org: String): Channel<FluxRecord>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream [FluxRecord]s to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream
+     * [com.influxdb.query.FluxRecord]s to [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
-     * @return the stream of [FluxRecord]s
+     * @return the stream of [com.influxdb.query.FluxRecord]s
      */
     fun query(query: Query): Channel<FluxRecord>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream [FluxRecord]s to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream
+     * [com.influxdb.query.FluxRecord]s to [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization
-     * @return the stream of [FluxRecord]s
+     * @return the stream of [com.influxdb.query.FluxRecord]s
      */
     fun query(query: Query, org: String): Channel<FluxRecord>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to
+     * [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
      * @param <M> the type of the measurement (POJO)
@@ -84,7 +89,8 @@ interface QueryKotlinApi {
     fun <M> query(query: String, measurementType: Class<M>): Channel<M>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to
+     * [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization
@@ -94,9 +100,10 @@ interface QueryKotlinApi {
     fun <M> query(query: String, org: String, measurementType: Class<M>): Channel<M>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to
+     * [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
      * @param <M> the type of the measurement (POJO)
@@ -105,7 +112,8 @@ interface QueryKotlinApi {
     fun <M> query(query: Query, measurementType: Class<M>): Channel<M>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream measurements to
+     * [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization
@@ -115,17 +123,19 @@ interface QueryKotlinApi {
     fun <M> query(query: Query, org: String, measurementType: Class<M>): Channel<M>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
      * @return the response stream
      */
     fun queryRaw(query: String): Channel<String>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization
@@ -134,9 +144,10 @@ interface QueryKotlinApi {
     fun queryRaw(query: String, org: String): Channel<String>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
      * @param dialect    Dialect is an object defining the options to use when encoding the response.
@@ -146,7 +157,8 @@ interface QueryKotlinApi {
     fun queryRaw(query: String, dialect: Dialect): Channel<String>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org  specifies the source organization
@@ -157,17 +169,19 @@ interface QueryKotlinApi {
     fun queryRaw(query: String, dialect: Dialect, org: String): Channel<String>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
-     * <p>The {@link InfluxDBClientOptions#getOrg()} will be used as source organization.</p>
+     * The [com.influxdb.client.InfluxDBClientOptions.getOrg] will be used as source organization.
      *
      * @param query the flux query to execute
      * @return the response stream
      */
     fun queryRaw(query: Query): Channel<String>
 
     /**
-     * Executes the Flux query against the InfluxDB and asynchronously stream response to [Channel].
+     * Executes the Flux query against the InfluxDB and asynchronously stream response to
+     * [kotlinx.coroutines.channels.Channel].
      *
      * @param query the flux query to execute
      * @param org specifies the source organization