// 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 static io.nats.client.Options.PROP_URL;
import java.io.IOException;
import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.spark.storage.StorageLevel;
import com.logimethods.connector.nats_spark.IncompleteException;
import io.nats.client.Connection;
import io.nats.client.Dispatcher;
import io.nats.client.MessageHandler;
import io.nats.client.Nats;
import io.nats.client.Options;

/*******************************************************************************
 * Copyright (c) 2016 Logimethods
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the MIT License (MIT)
 * which accompanies this distribution, and is available at
 * http://opensource.org/licenses/MIT
 *******************************************************************************/
package com.logimethods.connector.nats.to_spark;






/**
 * A NATS to Spark Connector.
 * <p>
 * It will transfer messages received from NATS into Spark data.
 * <p>
 * That class extends {@link com.logimethods.connector.nats.to_spark.NatsToSparkConnector}&lt;T,R,V&gt;.
 */
@SuppressWarnings("serial")
public abstract class OmnipotentStandardNatsToSparkConnector<T,R,V> extends NatsToSparkConnector<T,R,V> {
			
	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, Properties properties, StorageLevel storageLevel, String... subjects) {
		super(type, storageLevel, subjects);
		this.properties = properties;
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel, String... subjects) {
		super(type, storageLevel, subjects);
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, Properties properties, StorageLevel storageLevel) {
		super(type, storageLevel);
		this.properties = properties;
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel) {
		super(type, storageLevel);
		setNatsQueue();
	}
	
	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel, Collection<String> subjects, Properties properties, String queue, String natsUrl) {
		super(type, storageLevel, subjects, properties, queue, natsUrl);
	}

	/**
	 */
	protected StandardNatsToKeyValueSparkConnectorImpl<V> storedAsKeyValue() {
		return new StandardNatsToKeyValueSparkConnectorImpl<V>(type, storageLevel(), subjects, properties, natsQueue, natsUrl, dataDecoder, scalaDataDecoder);
	}

	protected Properties enrichedProperties;

	/** Create a socket connection and receive data until receiver is stopped 
	 * @throws IncompleteException 
	 * @throws TimeoutException 
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws IllegalArgumentException 
	 * @throws IllegalStateException **/

// 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/OmnipotentStandardNatsToSparkConnector.java
import static io.nats.client.Options.PROP_URL;
import java.io.IOException;
import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
import org.apache.spark.storage.StorageLevel;
import com.logimethods.connector.nats_spark.IncompleteException;
import io.nats.client.Connection;
import io.nats.client.Dispatcher;
import io.nats.client.MessageHandler;
import io.nats.client.Nats;
import io.nats.client.Options;

/*******************************************************************************
 * Copyright (c) 2016 Logimethods
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the MIT License (MIT)
 * which accompanies this distribution, and is available at
 * http://opensource.org/licenses/MIT
 *******************************************************************************/
package com.logimethods.connector.nats.to_spark;






/**
 * A NATS to Spark Connector.
 * <p>
 * It will transfer messages received from NATS into Spark data.
 * <p>
 * That class extends {@link com.logimethods.connector.nats.to_spark.NatsToSparkConnector}&lt;T,R,V&gt;.
 */
@SuppressWarnings("serial")
public abstract class OmnipotentStandardNatsToSparkConnector<T,R,V> extends NatsToSparkConnector<T,R,V> {
			
	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, Properties properties, StorageLevel storageLevel, String... subjects) {
		super(type, storageLevel, subjects);
		this.properties = properties;
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel, String... subjects) {
		super(type, storageLevel, subjects);
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, Properties properties, StorageLevel storageLevel) {
		super(type, storageLevel);
		this.properties = properties;
		setNatsQueue();
	}

	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel) {
		super(type, storageLevel);
		setNatsQueue();
	}
	
	protected OmnipotentStandardNatsToSparkConnector(Class<V> type, StorageLevel storageLevel, Collection<String> subjects, Properties properties, String queue, String natsUrl) {
		super(type, storageLevel, subjects, properties, queue, natsUrl);
	}

	/**
	 */
	protected StandardNatsToKeyValueSparkConnectorImpl<V> storedAsKeyValue() {
		return new StandardNatsToKeyValueSparkConnectorImpl<V>(type, storageLevel(), subjects, properties, natsQueue, natsUrl, dataDecoder, scalaDataDecoder);
	}

	protected Properties enrichedProperties;

	/** Create a socket connection and receive data until receiver is stopped 
	 * @throws IncompleteException 
	 * @throws TimeoutException 
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws IllegalArgumentException 
	 * @throws IllegalStateException **/

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