flink-connector-influxdb2/src/main/java/org/apache/flink/streaming/connectors/influxdb/sink/InfluxDBSinkOptions.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.flink.streaming.connectors.influxdb.sink;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.InfluxDBClientOptions;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;

public final class InfluxDBSinkOptions {

    private InfluxDBSinkOptions() {}

    public static final ConfigOption<Boolean> WRITE_DATA_POINT_CHECKPOINT =
            ConfigOptions.key("sink.influxDB.write.data_point.checkpoint")
                    .booleanType()
                    .defaultValue(false)
                    .withDescription(
                            "Determines if the checkpoint data point should be written to InfluxDB or not.");

    public static final ConfigOption<Integer> WRITE_BUFFER_SIZE =
            ConfigOptions.key("sink.influxDB.write.buffer.size")
                    .intType()
                    .defaultValue(1000)
                    .withDescription(
                            "Size of the buffer to store the data before writing to InfluxDB.");

    public static final ConfigOption<String> INFLUXDB_URL =
            ConfigOptions.key("sink.influxDB.client.URL")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB Connection URL.");

    public static final ConfigOption<String> INFLUXDB_USERNAME =
            ConfigOptions.key("sink.influxDB.client.username")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB username.");

    public static final ConfigOption<String> INFLUXDB_PASSWORD =
            ConfigOptions.key("sink.influxDB.client.password")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB password.");

    public static final ConfigOption<String> INFLUXDB_TOKEN =
            ConfigOptions.key("sink.influxDB.client.token")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB access token.");

    public static final ConfigOption<String> INFLUXDB_BUCKET =
            ConfigOptions.key("sink.influxDB.client.bucket")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB bucket name.");

    public static final ConfigOption<String> INFLUXDB_ORGANIZATION =
            ConfigOptions.key("sink.influxDB.client.organization")
                    .stringType()
                    .noDefaultValue()
                    .withDescription("InfluxDB organization name.");

    public static InfluxDBClient getInfluxDBClient(final Configuration configuration) {
        final String url = configuration.getString(INFLUXDB_URL);
        final String username = configuration.getString(INFLUXDB_USERNAME);
        final String password = configuration.getString(INFLUXDB_PASSWORD);
        final String token = configuration.getString(INFLUXDB_TOKEN);
        final String bucket = configuration.getString(INFLUXDB_BUCKET);
        final String organization = configuration.getString(INFLUXDB_ORGANIZATION);
        InfluxDBClientOptions.Builder builder = InfluxDBClientOptions.builder();
        builder = builder
                .url(url)
                .bucket(bucket)
                .org(organization);
        if (token != null) {
            builder = builder.authenticateToken(token.toCharArray());
        } else if (username != null && password != null) {
            builder = builder.authenticate(username, password.toCharArray());
        }
        final InfluxDBClientOptions influxDBClientOptions = builder.build();
        return InfluxDBClientFactory.create(influxDBClientOptions);
    }
}