<dependencies>
    <dependency>
        <groupId>com.here.platform.data.client</groupId>
        <artifactId>flink-support_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.pipeline</groupId>
        <artifactId>pipeline-interface_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
    </dependency>
</dependencies>

/*
 * Copyright (c) 2018-2022 HERE Europe B.V.
 *
 * Licensed 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.
 */

import com.here.hrn.HRN;
import com.here.platform.data.client.flink.javadsl.*;
import com.here.platform.data.client.javadsl.CommitPartition;
import com.here.platform.data.client.javadsl.NewPartition;
import com.here.platform.data.client.javadsl.Partition;
import com.here.platform.data.client.settings.ConsumerSettings;
import com.here.platform.pipeline.PipelineContext;
import java.io.Serializable;
import java.util.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

// Uploads partitions to blobstore.
class UploadDataFunction extends RichMapFunction<Partition, CommitPartition>
    implements Serializable {
  private final HRN sensorDataHRN;
  private final HRN outputHRN;
  private transient FlinkDataClient dataClient;
  private transient FlinkReadEngine readEngine;
  private transient FlinkWriteEngine writeEngine;

  public UploadDataFunction(HRN sensorDataHRN, HRN outputHRN) {
    this.sensorDataHRN = sensorDataHRN;
    this.outputHRN = outputHRN;
  }

  @Override
  public void open(Configuration parameters) {
    dataClient = new FlinkDataClient();
    readEngine = dataClient.readEngine(sensorDataHRN);
    writeEngine = dataClient.writeEngine(outputHRN);
  }

  @Override
  public void close() {
    dataClient.terminate();
  }

  @Override
  public CommitPartition map(Partition partition) {
    byte[] data = readEngine.getDataAsBytes(partition);
    NewPartition newPartition =
        new NewPartition.Builder()
            .withPartition(partition.getPartition())
            .withLayer("sdii-message-archive")
            .withData(data)
            .build();
    return writeEngine.put(newPartition);
  }
}

// Window function that publishes a new batch version for all CommitPartitions in the window.
class PublishBatchWindowFunction extends RichAllWindowFunction<CommitPartition, String, TimeWindow>
    implements Serializable {
  private final HRN hrn;
  private transient FlinkDataClient dataClient;
  private transient FlinkPublishApi publishApi;

  public PublishBatchWindowFunction(HRN hrn) {
    this.hrn = hrn;
  }

  @Override
  public void open(Configuration parameters) {
    dataClient = new FlinkDataClient();
    publishApi = dataClient.publishApi(hrn);
  }

  @Override
  public void close() {
    dataClient.terminate();
  }

  @Override
  public void apply(
      TimeWindow window, Iterable<CommitPartition> commitPartitions, Collector<String> out) {
    OptionalLong baseVersion = publishApi.getBaseVersion();

    Set<String> layersId = new HashSet<>();
    commitPartitions.forEach(partition -> layersId.add(partition.getLayer()));

    // Note that we do not log any versioned dependencies, because we are only processing streaming
    // input.
    publishApi.publishBatch2(
        baseVersion,
        Optional.of(new ArrayList<>(layersId)),
        Collections.emptyList(),
        commitPartitions.iterator());
    out.collect("commit on " + baseVersion + " success");
  }
}

public class ArchiveStreamJava {

  public static void main(String[] args) {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    PipelineContext pipelineContext = new PipelineContext();
    HRN sensorDataHRN = pipelineContext.getConfig().getInputCatalogs().get("sensorData");
    HRN outputHRN = pipelineContext.getConfig().getOutputCatalog();

    String streamingLayer = "sample-streaming-layer";

    FlinkDataClient flinkDataClient = new FlinkDataClient();

    FlinkQueryApi queryApi = flinkDataClient.queryApi(sensorDataHRN);

    ConsumerSettings consumerSettings =
        new ConsumerSettings.Builder()
            .withLatestOffset()
            .withGroupName("archive-stream-consumer")
            .build();

    SourceFunction<Partition> subscriptionFunction =
        queryApi.subscribe(streamingLayer, consumerSettings);

    // Subscribe to streaming sensor data messages.
    DataStream<Partition> partitions = env.addSource(subscriptionFunction);

    try {
      partitions
          // Add periodic watermark generation
          .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps())
          // Upload incoming streaming sensor data messages to the blobstore for the versioned
          // layer.
          .map(new UploadDataFunction(sensorDataHRN, outputHRN))

          // Publishes a new layer version once per minute.
          .windowAll(TumblingProcessingTimeWindows.of(Time.minutes(1)))
          .apply(new PublishBatchWindowFunction(outputHRN))
          .addSink(
              new SinkFunction<String>() {
                @Override
                public void invoke(String value, Context context) throws Exception {
                  /// [halt-processing]
                  // If you comment out this line, the application will continue to publish a new
                  // version to the output catalog once per minute, until you explicitly kill
                  // the application if running locally, or cancel the pipeline version if
                  // running on a pipeline.
                  throw new InterruptedException("Halting stream processing");
                  /// [halt-processing]
                }
              });

      env.execute();
    } catch (Exception ignored) {
    }

    subscriptionFunction.cancel();
    flinkDataClient.terminate();
  }
}

mvn compile exec:java -Dexec.mainClass=ArchiveStreamJava -Dpipeline-config.file=pipeline-config.conf \
        -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}
        

// If you comment out this line, the application will continue to publish a new
// version to the output catalog once per minute, until you explicitly kill
// the application if running locally, or cancel the pipeline version if
// running on a pipeline.
throw new InterruptedException("Halting stream processing");