// Path: src/main/java/com/logimethods/connector/nats_spark/IncompleteException.java
// public class IncompleteException extends Exception {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1L;
// 
// 	public IncompleteException() {
// 	}
// 
// 	public IncompleteException(String message) {
// 		super(message);
// 	}
// 
// 	public IncompleteException(Throwable cause) {
// 		super(cause);
// 	}
// 
// 	public IncompleteException(String message, Throwable cause) {
// 		super(message, cause);
// 	}
// 
// 	public IncompleteException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
// 		super(message, cause, enableSuppression, writableStackTrace);
// 	}
// 
// }


import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.logimethods.connector.nats_spark.IncompleteException;
import io.nats.streaming.Message;
import io.nats.streaming.MessageHandler;
import io.nats.streaming.NatsStreaming;
import io.nats.streaming.Options;
import io.nats.streaming.Subscription;
import io.nats.streaming.SubscriptionOptions;

	/**
	 * @return the opts
	 */
	protected SubscriptionOptions getSubscriptionOptions() {
		if (subscriptionOpts == null){
			subscriptionOpts = getSubscriptionOptsBuilder().dispatcher(DISPATCHER_NAME).build();
		}
		return subscriptionOpts;
	}

	/**
	 * @return the optsBuilder
	 */
	protected SubscriptionOptions.Builder getSubscriptionOptsBuilder() {
		if (subscriptionOptsBuilder == null) {
			subscriptionOptsBuilder = new SubscriptionOptions.Builder();
		}
		return subscriptionOptsBuilder;
	}

	/**
	 * @return a NATS Streaming to Spark Connector where the NATS Messages are stored in Spark as Key (the NATS Subject) / Value (the NATS Payload)
	 */
	public NatsStreamingToKeyValueSparkConnectorImpl<V> storedAsKeyValue() {
		return new NatsStreamingToKeyValueSparkConnectorImpl<V>(type, storageLevel(), subjects, properties, natsQueue, natsUrl, clusterID, clientID, 
																subscriptionOpts, subscriptionOptsBuilder, dataDecoder, scalaDataDecoder);
	}

	/** Create a socket connection and receive data until receiver is stopped 
	 * @throws Exception **/

// Path: src/main/java/com/logimethods/connector/nats_spark/IncompleteException.java
// public class IncompleteException extends Exception {
// 
// 	/**
// 	 * 
// 	 */
// 	private static final long serialVersionUID = 1L;
// 
// 	public IncompleteException() {
// 	}
// 
// 	public IncompleteException(String message) {
// 		super(message);
// 	}
// 
// 	public IncompleteException(Throwable cause) {
// 		super(cause);
// 	}
// 
// 	public IncompleteException(String message, Throwable cause) {
// 		super(message, cause);
// 	}
// 
// 	public IncompleteException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
// 		super(message, cause, enableSuppression, writableStackTrace);
// 	}
// 
// }


// Path: src/main/java/com/logimethods/connector/nats/to_spark/OmnipotentNatsStreamingToSparkConnector.java
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.logimethods.connector.nats_spark.IncompleteException;
import io.nats.streaming.Message;
import io.nats.streaming.MessageHandler;
import io.nats.streaming.NatsStreaming;
import io.nats.streaming.Options;
import io.nats.streaming.Subscription;
import io.nats.streaming.SubscriptionOptions;

	/**
	 * @return the opts
	 */
	protected SubscriptionOptions getSubscriptionOptions() {
		if (subscriptionOpts == null){
			subscriptionOpts = getSubscriptionOptsBuilder().dispatcher(DISPATCHER_NAME).build();
		}
		return subscriptionOpts;
	}

	/**
	 * @return the optsBuilder
	 */
	protected SubscriptionOptions.Builder getSubscriptionOptsBuilder() {
		if (subscriptionOptsBuilder == null) {
			subscriptionOptsBuilder = new SubscriptionOptions.Builder();
		}
		return subscriptionOptsBuilder;
	}

	/**
	 * @return a NATS Streaming to Spark Connector where the NATS Messages are stored in Spark as Key (the NATS Subject) / Value (the NATS Payload)
	 */
	public NatsStreamingToKeyValueSparkConnectorImpl<V> storedAsKeyValue() {
		return new NatsStreamingToKeyValueSparkConnectorImpl<V>(type, storageLevel(), subjects, properties, natsQueue, natsUrl, clusterID, clientID, 
																subscriptionOpts, subscriptionOptsBuilder, dataDecoder, scalaDataDecoder);
	}

	/** Create a socket connection and receive data until receiver is stopped 
	 * @throws Exception **/

	protected void receive() throws IOException, InterruptedException, IncompleteException, TimeoutException {
