/*
 * 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.cassandra.example;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.cassandra.CassandraSink;
import org.apache.flink.streaming.connectors.cassandra.ClusterBuilder;

import com.datastax.driver.core.Cluster;

import java.util.Collections;
import java.util.List;
import java.util.UUID;

/**
 * This is an example showing the to use the Cassandra Sink (with write-ahead log) in the Streaming API.
 *
 * <p>The example assumes that a table exists in a local cassandra database, according to the following queries:
 * CREATE KEYSPACE IF NOT EXISTS example WITH replication = {'class': 'SimpleStrategy', 'replication_factor': ‘1’};
 * CREATE TABLE example.values (id text, count int, PRIMARY KEY(id));
 *
 * <p>Important things to note are that checkpointing is enabled, a StateBackend is set and the enableWriteAheadLog() call
 * when creating the CassandraSink.
 */
public class CassandraTupleWriteAheadSinkExample {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		env.enableCheckpointing(1000);
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 1000));
		env.setStateBackend(new FsStateBackend("file:///" + System.getProperty("java.io.tmpdir") + "/flink/backend"));

		CassandraSink<Tuple2<String, Integer>> sink = CassandraSink.addSink(env.addSource(new MySource()))
			.setQuery("INSERT INTO example.values (id, counter) values (?, ?);")
			.enableWriteAheadLog()
			.setClusterBuilder(new ClusterBuilder() {

				private static final long serialVersionUID = 2793938419775311824L;

				@Override
				public Cluster buildCluster(Cluster.Builder builder) {
					return builder.addContactPoint("127.0.0.1").build();
				}
			})
			.build();

		sink.name("Cassandra Sink").disableChaining().setParallelism(1).uid("hello");

		env.execute();
	}

	private static class MySource implements SourceFunction<Tuple2<String, Integer>>, ListCheckpointed<Integer> {
		private static final long serialVersionUID = 4022367939215095610L;

		private int counter = 0;
		private boolean stop = false;

		@Override
		public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
			while (!stop) {
				Thread.sleep(50);
				ctx.collect(new Tuple2<>("" + UUID.randomUUID(), 1));
				counter++;
				if (counter == 100) {
					stop = true;
				}
			}
		}

		@Override
		public void cancel() {
			stop = true;
		}

		@Override
		public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
			return Collections.singletonList(this.counter);
		}

		@Override
		public void restoreState(List<Integer> state) throws Exception {
			if (state.isEmpty() || state.size() > 1) {
				throw new RuntimeException("Test failed due to unexpected recovered state size " + state.size());
			}
			this.counter = state.get(0);
		}
	}
}