• influxdb-client-kotlin
• Documentation
• Features
• Queries
• Writes
• Advanced Usage
  • Client configuration file
  • Configuration example
• Client connection string
• Gzip support
• Log HTTP Request and Response
• Check the server status
• Construct queries using the flux-dsl query builder
• Version
• Snapshot Repository
  • Maven
  • Gradle

Kotlin - 图1influxdb-client-kotlin

The reference Kotlin client that allows query and write for the InfluxDB 2.x by Kotlin Channel coroutines.

Kotlin - 图3Documentation

This section contains links to the client library documentation.

• Product documentation, Getting Started
• Examples
• API Reference
• Changelog

Kotlin - 图4Features

• Querying data using Flux language
• Writing data
• Advanced Usage

Kotlin - 图5Queries

The QueryKotlinApi supports asynchronous queries by Kotlin Channel coroutines.

The following example demonstrates querying using the Flux language:

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import kotlinx.coroutines.channels.consumeEach
5. import kotlinx.coroutines.channels.filter
6. import kotlinx.coroutines.channels.take
7. import kotlinx.coroutines.runBlocking

9. fun main(args: Array<String>) = runBlocking {

11. val influxDBClient = InfluxDBClientKotlinFactory
12. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

14. val fluxQuery = ("from(bucket: \"my-bucket\")\n"
15. + " |> range(start: -1d)"
16. + " |> filter(fn: (r) => (r[\"_measurement\"] == \"cpu\" and r[\"_field\"] == \"usage_system\"))")

18. //Result is returned as a stream
19. val results = influxDBClient.getQueryKotlinApi().query(fluxQuery)

21. //Example of additional result stream processing on client side
22. results
23. //filter on client side using `filter` built-in operator
24. .filter { "cpu0" == it.getValueByKey("cpu") }
25. //take first 20 records
26. .take(20)
27. //print results
28. .consumeEach { println("Measurement: ${it.measurement}, value: ${it.value}") }

30. influxDBClient.close()
31. }

It is possible to parse a result line-by-line using the queryRaw method:

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import kotlinx.coroutines.channels.consumeEach
5. import kotlinx.coroutines.runBlocking

7. fun main(args: Array<String>) = runBlocking {

9. val influxDBClient = InfluxDBClientKotlinFactory
10. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

12. val fluxQuery = ("from(bucket: \"my-bucket\")\n"
13. + " |> range(start: -5m)"
14. + " |> filter(fn: (r) => (r[\"_measurement\"] == \"cpu\" and r[\"_field\"] == \"usage_system\"))"
15. + " |> sample(n: 5, pos: 1)")

17. //Result is returned as a stream
18. val results = influxDBClient.getQueryKotlinApi().queryRaw(fluxQuery)

20. //print results
21. results.consumeEach { println("Line: $it") }

23. influxDBClient.close()
24. }

Kotlin - 图6Writes

The WriteKotlinApi supports ingest data by:

• DataPoint
• LineProtocol
• Data class
• List of above items

The following example shows how to use various type of data:

1. package example

3. import com.influxdb.annotations.Column
4. import com.influxdb.annotations.Measurement
5. import com.influxdb.client.domain.WritePrecision
6. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
7. import com.influxdb.client.write.Point
8. import kotlinx.coroutines.flow.collect
9. import kotlinx.coroutines.flow.consumeAsFlow
10. import kotlinx.coroutines.runBlocking
11. import java.time.Instant

13. fun main() = runBlocking {

15. val org = "my-org"
16. val bucket = "my-bucket"

18. //
19. // Initialize client
20. //
21. val client = InfluxDBClientKotlinFactory
22. .create("http://localhost:8086", "my-token".toCharArray(), org, bucket)

24. val writeApi = client.getWriteKotlinApi()

26. //
27. // Write by Data Point
28. //
29. val point = Point.measurement("temperature")
30. .addTag("location", "west")
31. .addField("value", 55.0)
32. .time(Instant.now().toEpochMilli(), WritePrecision.MS)

34. writeApi.writePoint(point)

36. //
37. // Write by LineProtocol
38. //
39. writeApi.writeRecord("temperature,location=north value=60.0", WritePrecision.NS)

41. //
42. // Write by DataClass
43. //
44. val temperature = Temperature("south", 62.0, Instant.now())

46. writeApi.writeMeasurement(temperature, WritePrecision.NS)

48. //
49. // Query results
50. //
51. val fluxQuery =
52. """from(bucket: "$bucket") |> range(start: 0) |> filter(fn: (r) => (r["_measurement"] == "temperature"))"""

54. client
55. .getQueryKotlinApi()
56. .query(fluxQuery)
57. .consumeAsFlow()
58. .collect { println("Measurement: ${it.measurement}, value: ${it.value}") }

60. client.close()
61. }

63. @Measurement(name = "temperature")
64. data class Temperature(
65. @Column(tag = true) val location: String,
66. @Column val value: Double,
67. @Column(timestamp = true) val time: Instant
68. )

• sources - KotlinWriteApi.kt

Kotlin - 图7Advanced Usage

Kotlin - 图8Client configuration file

A client can be configured via configuration file. The configuration file has to be named as influx2.properties and has to be in root of classpath.

The following options are supported:

The influx2.readTimeout, influx2.writeTimeout and influx2.connectTimeout supports ms, s and m as unit. Default is milliseconds.

Kotlin - 图9Configuration example

1. influx2.url=http://localhost:8086
2. influx2.org=my-org
3. influx2.bucket=my-bucket
4. influx2.token=my-token
5. influx2.logLevel=BODY
6. influx2.readTimeout=5s
7. influx2.writeTimeout=10s
8. influx2.connectTimeout=5s

and then:

1. val influxDBClient = InfluxDBClientKotlinFactory.create();

Kotlin - 图10Client connection string

A client can be constructed using a connection string that can contain the InfluxDBClientOptions parameters encoded into the URL.

1. val influxDBClient = InfluxDBClientKotlinFactory
2. .create("http://localhost:8086?readTimeout=5000&connectTimeout=5000&logLevel=BASIC", token)

The following options are supported:

The readTimeout, writeTimeout and connectTimeout supports ms, s and m as unit. Default is milliseconds.

Kotlin - 图11Gzip support

InfluxDBClientKotlin does not enable gzip compress for http requests by default. If you want to enable gzip to reduce transfer data’s size, you can call:

1. influxDBClient.enableGzip();

Kotlin - 图12Log HTTP Request and Response

The Requests and Responses can be logged by changing the LogLevel. LogLevel values are NONE, BASIC, HEADER, BODY. Note that applying the BODY LogLevel will disable chunking while streaming and will load the whole response into memory.

1. influxDBClient.setLogLevel(LogLevel.HEADERS)

Kotlin - 图13Check the server status

Server availability can be checked using the influxDBClient.ping() endpoint.

Kotlin - 图14Construct queries using the flux-dsl query builder

1. package example

3. import com.influxdb.client.kotlin.InfluxDBClientKotlinFactory
4. import com.influxdb.query.dsl.Flux
5. import com.influxdb.query.dsl.functions.restriction.Restrictions
6. import kotlinx.coroutines.channels.consumeEach
7. import kotlinx.coroutines.channels.filter
8. import kotlinx.coroutines.channels.take
9. import kotlinx.coroutines.runBlocking
10. import java.time.temporal.ChronoUnit

12. fun main(args: Array<String>) = runBlocking {

14. val influxDBClient = InfluxDBClientKotlinFactory
15. .create("http://localhost:8086", "my-token".toCharArray(), "my-org")

17. val mem = Flux.from("my-bucket")
18. .range(-30L, ChronoUnit.MINUTES)
19. .filter(Restrictions.and(Restrictions.measurement().equal("mem"), Restrictions.field().equal("used_percent")))

21. //Result is returned as a stream
22. val results = influxDBClient.getQueryKotlinApi().query(mem.toString())

24. //Example of additional result stream processing on client side
25. results
26. //filter on client side using `filter` built-in operator
27. .filter { (it.value as Double) > 55 }
28. // take first 20 records
29. .take(20)
30. //print results
31. .consumeEach { println("Measurement: ${it.measurement}, value: ${it.value}") }

33. influxDBClient.close()
34. }

Kotlin - 图15Version

The latest version for Maven dependency:

1. <dependency>
2. <groupId>com.influxdb</groupId>
3. <artifactId>influxdb-client-kotlin</artifactId>
4. <version>6.9.0</version>
5. </dependency>

Or when using with Gradle:

1. dependencies {
2. implementation "com.influxdb:influxdb-client-kotlin:6.9.0"
3. }

Kotlin - 图16Snapshot Repository

The snapshots are deployed into OSS Snapshot repository.

Kotlin - 图17Maven

1. <repository>
2. <id>ossrh</id>
3. <name>OSS Snapshot repository</name>
4. <url>https://oss.sonatype.org/content/repositories/snapshots/</url>
5. <releases>
6. <enabled>false</enabled>
7. </releases>
8. <snapshots>
9. <enabled>true</enabled>
10. </snapshots>
11. </repository>

Kotlin - 图18Gradle

1. repositories {
2. maven { url "https://oss.sonatype.org/content/repositories/snapshots" }
3. }