PowerBIPush.java

/* Generated by Streams Studio: January 1, 2016 at 11:12:21 PM GMT+1 */
package com.ibm.streams.powerbi;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.client.ClientProtocolException;
import org.apache.log4j.Logger;
import org.com.powerbi.streams.PowerBI;
import org.com.powerbi.streams.PowerBI.TableValue;

import com.ibm.streams.operator.AbstractOperator;
import com.ibm.streams.operator.OperatorContext;
import com.ibm.streams.operator.StreamSchema;
import com.ibm.streams.operator.StreamingData.Punctuation;
import com.ibm.streams.operator.StreamingInput;
import com.ibm.streams.operator.Tuple;
import com.ibm.streams.operator.Type;
import com.ibm.streams.operator.log4j.TraceLevel;
import com.ibm.streams.operator.model.InputPortSet;
import com.ibm.streams.operator.model.InputPortSet.WindowMode;
import com.ibm.streams.operator.model.InputPortSet.WindowPunctuationInputMode;
import com.ibm.streams.operator.model.InputPorts;
import com.ibm.streams.operator.model.Libraries;
import com.ibm.streams.operator.model.Parameter;
import com.ibm.streams.operator.model.PrimitiveOperator;

/**
 * Class for an operator that consumes tuples and does not produce an output
 * stream. This pattern supports a number of input streams and no output
 * streams.
 * <P>
 * The following event methods from the Operator interface can be called:
 * </p>
 * <ul>
 * <li><code>initialize()</code> to perform operator initialization</li>
 * <li>allPortsReady() notification indicates the operator's ports are ready to
 * process and submit tuples</li>
 * <li>process() handles a tuple arriving on an input port
 * <li>processPuncuation() handles a punctuation mark arriving on an input port
 * <li>shutdown() to shutdown the operator. A shutdown request may occur at any
 * time, such as a request to stop a PE or cancel a job. Thus the shutdown() may
 * occur while the operator is processing tuples, punctuation marks, or even
 * during port ready notification.</li>
 * </ul>
 * <p>
 * With the exception of operator initialization, all the other events may occur
 * concurrently with each other, which lead to these methods being called
 * concurrently by different threads.
 * </p>
 */
@Libraries({ "impl/lib/*" })
@PrimitiveOperator(name = "PowerBI", namespace = "com.ibm.streams.powerbi", description = "Java Operator PowerBI")
@InputPorts({
		@InputPortSet(description = "Port that ingests tuples", cardinality = 1, optional = false, windowingMode = WindowMode.NonWindowed, windowPunctuationInputMode = WindowPunctuationInputMode.Oblivious) })
public class PowerBIPush extends AbstractOperator {

	private final Logger log = Logger.getLogger(this.getClass());

	/** parameter */
	private String oauth_username;
	/** parameter */
	private String oauth_password;
	/** parameter */
	private String oauth_clientid;
	/** parameter */
	private String datasetName;
	/** parameter */
	private String tablename;
	/** parameter */
	private int flushsize = 1;
	/** parameter */
	private boolean clearfirstly = false;

	/** token id after authentication */
	private String token = null;
	/** data set id */
	private String datasetid = null;
	/** buffer for tuples. Access if synchronized. */
	private final List<Map<String, TableValue>> rlist = new ArrayList<Map<String, TableValue>>();

	private boolean isPunctuation() {
		return flushsize == 0;
	}

	public static class PowerException extends Exception {

		private static final long serialVersionUID = 1L;

		PowerException(String mess) {
			super(mess);
		}
	}

	private void failure(String mess) throws PowerException {
		log.log(TraceLevel.ERROR, mess);
		throw new PowerException(mess);
	}

	private Map<String, String> createTableSchema(OperatorContext context) throws PowerException {
		StreamSchema sche = context.getStreamingInputs().get(0).getStreamSchema();
		Map<String, String> bischema = new HashMap<String, String>();
		for (String name : sche.getAttributeNames()) {
			Type.MetaType ty = sche.getAttribute(name).getType().getMetaType();
			String biType = null;
			switch (ty) {
			case RSTRING:
			case USTRING:
			case ENUM:
				biType = PowerBI.STRING_TYPE;
				break;
			case INT8:
			case INT32:
			case INT64:
			case UINT8:
			case UINT16:
			case UINT32:
			case UINT64:
			case INT16:
				biType = PowerBI.INT64_TYPE;
				break;
			case DECIMAL128:
			case DECIMAL32:
			case DECIMAL64:
			case FLOAT32:
			case FLOAT64:
				biType = PowerBI.DOUBLE_TYPE;
				break;
			case BOOLEAN:
				biType = PowerBI.BOOL_TYPE;
				break;
			case TIMESTAMP:
				biType = PowerBI.DATETIME_TYPE;
				break;
			default:
				break;
			}
			if (biType == null)
				failure("Attribute " + name + " type " + ty.getLanguageType() + " not supported");
			log.log(TraceLevel.DEBUG, "Attribute " + name + " type " + ty.getLanguageType() + " mapped to " + biType);
			bischema.put(name, biType);
		}
		return bischema;
	}

	/**
	 * Initialize this operator. Called once before any tuples are processed.
	 * 
	 * @param context
	 *            OperatorContext for this operator.
	 * @throws Exception
	 *             Operator failure, will cause the enclosing PE to terminate.
	 */
	@Override
	public synchronized void initialize(OperatorContext context) throws Exception {
		// Must call super.initialize(context) to correctly setup an operator.
		super.initialize(context);
		log.trace("Operator " + context.getName() + " initializing in PE: " + context.getPE().getPEId() + " in Job: "
				+ context.getPE().getJobId());
		log.log(TraceLevel.INFO, "Logging on the Power BI using " + oauth_username);
		log.log(TraceLevel.INFO, "Database set " + datasetName + " table name " + tablename);
		if (clearfirstly)
			log.log(TraceLevel.INFO, "Data is cleared at the beginning");
		else
			log.log(TraceLevel.INFO, "Data is not cleared at the beginning");
		log.log(TraceLevel.INFO, "Flush buffer size " + flushsize);
		if (isPunctuation())
			log.log(TraceLevel.INFO, "Data is flushed and sent to PowerBI at the punctuation marker.");
		token = PowerBI.getAuthToken(oauth_username, oauth_password, oauth_clientid);
		if (token == null)
			failure("Authentication failed");
		log.log(TraceLevel.INFO, "Authentication successfull");

		log.log(TraceLevel.INFO, "Check or create table schema if necessary");
		Map<String, String> schema = createTableSchema(context);
		datasetid = PowerBI.checkTableDataSet(token, datasetName, tablename, schema);
		if (clearfirstly) {
			log.log(TraceLevel.INFO, "Data cleansing started");
			PowerBI.clearTable(token, datasetid, tablename);
			log.log(TraceLevel.INFO, "Data cleansing completed");
		}
		if (datasetid == null)
			failure("Cannot find or create table " + tablename + " in data set " + datasetName);
		log.log(TraceLevel.INFO, "Success, table name " + tablename + " in " + datasetName + " accessible");
	}

	/**
	 * Notification that initialization is complete and all input and output
	 * ports are connected and ready to receive and submit tuples.
	 * 
	 * @throws Exception
	 *             Operator failure, will cause the enclosing PE to terminate.
	 */
	@Override
	public synchronized void allPortsReady() throws Exception {
		// This method is commonly used by source operators.
		// Operators that process incoming tuples generally do not need this
		// notification.
		OperatorContext context = getOperatorContext();
		log.trace("Operator " + context.getName() + " all ports are ready in PE: " + context.getPE().getPEId()
				+ " in Job: " + context.getPE().getJobId());
	}

	private synchronized void pushData(Map<String, TableValue> elem) throws ClientProtocolException, IOException {
		rlist.add(elem);
		if (rlist.size() >= flushsize && !isPunctuation()) {
			PowerBI.addTableRows(token, datasetid, tablename, rlist);
			rlist.clear();
		}
	}

	private synchronized void pushPunctuation() throws ClientProtocolException, IOException {
		PowerBI.addTableRows(token, datasetid, tablename, rlist);
		rlist.clear();
	}

	/**
	 * Process an incoming tuple that arrived on the specified port.
	 * 
	 * @param stream
	 *            Port the tuple is arriving on.
	 * @param tuple
	 *            Object representing the incoming tuple.
	 * @throws Exception
	 *             Operator failure, will cause the enclosing PE to terminate.
	 */
	@Override
	public void process(StreamingInput<Tuple> stream, Tuple tuple) throws Exception {
		StreamSchema sche = stream.getStreamSchema();
		Map<String, TableValue> elem = new HashMap<String, TableValue>();
		for (String name : sche.getAttributeNames()) {
			Type.MetaType ty = sche.getAttribute(name).getType().getMetaType();
			TableValue val = null;
			switch (ty) {
			case RSTRING:
			case USTRING:
			case ENUM:
				val = new TableValue(tuple.getString(name));
				break;
			case INT8:
			case INT32:
			case INT64:
			case INT16:
			case UINT8:
			case UINT16:
			case UINT32:
			case UINT64:
				val = new TableValue(tuple.getLong(name));
				break;
			case DECIMAL128:
			case DECIMAL32:
			case DECIMAL64:
			case FLOAT32:
			case FLOAT64:
				val = new TableValue(tuple.getDouble(name));
				break;
			case BOOLEAN:
				val = new TableValue(tuple.getBoolean(name));
				break;
			case TIMESTAMP:
				val = new TableValue(new Date(tuple.getTimestamp(name).getTime()), null);
				break;
			default:
				break;
			}
			if (val != null)
				elem.put(name, val);
		}
		pushData(elem);
	}

	/**
	 * Process an incoming punctuation that arrived on the specified port.
	 * 
	 * @param stream
	 *            Port the punctuation is arriving on.
	 * @param mark
	 *            The punctuation mark
	 * @throws Exception
	 *             Operator failure, will cause the enclosing PE to terminate.
	 */
	@Override
	public void processPunctuation(StreamingInput<Tuple> stream, Punctuation mark) throws Exception {
		if (isPunctuation())
			pushPunctuation();
	}

	/**
	 * Shutdown this operator.
	 * 
	 * @throws Exception
	 *             Operator failure, will cause the enclosing PE to terminate.
	 */
	@Override
	public synchronized void shutdown() throws Exception {
		OperatorContext context = getOperatorContext();
		log.trace("Operator " + context.getName() + " shutting down in PE: " + context.getPE().getPEId() + " in Job: "
				+ context.getPE().getJobId());

		// TODO: If needed, close connections or release resources related to
		// any external system or data store.

		// Must call super.shutdown()
		super.shutdown();
	}

	@Parameter(description = "This parameter is mandatory. It is oauth username.")
	public void setOauth_username(String oauth_username) {
		this.oauth_username = oauth_username;
	}

	@Parameter(description = "This parameter is mandatory. It is oauth password.")
	public void setOauth_password(String oauth_password) {
		this.oauth_password = oauth_password;
	}

	@Parameter(description = "This parameter is mandatory. It is AZURE AD client id.")
	public void setOauth_clientid(String oauth_clientid) {
		this.oauth_clientid = oauth_clientid;
	}

	@Parameter(description = "This parameter is mandatory. It is table name inside data set.")
	public void setTablename(String tablename) {
		this.tablename = tablename;
	}

	@Parameter(description = "This parameter is mandatory. It is data set name.")
	public void setDatasetName(String datasetName) {
		this.datasetName = datasetName;
	}

	@Parameter(optional = true, description = "This parameter is optional. Number of tuples to be sent to PowerBI. If not specified default is 1. If 0 then data is flushed at the puctuation  marker.")
	public void setFlushsize(int flushsize) {
		this.flushsize = flushsize;
	}

	@Parameter(optional = true, description = "This parameter is optional. If true then table is clear at the begining.")
	public void setClearfirstly(boolean clearfirstly) {
		this.clearfirstly = clearfirstly;
	}

}