//*******************************************************************************
//* Copyright (C) 2017 International Business Machines Corporation
//* All Rights Reserved
//*******************************************************************************

package com.ibm.streamsx.health.microservices;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.log4j.Logger;

import com.ibm.streams.operator.logging.TraceLevel;
import com.ibm.streamsx.topology.Topology;
import com.ibm.streamsx.topology.context.ContextProperties;
import com.ibm.streamsx.topology.context.JobProperties;
import com.ibm.streamsx.topology.context.StreamsContextFactory;

public abstract class AbstractService implements Serializable {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	private Properties properties;

	protected static Logger TRACE = Logger.getLogger(AbstractService.class);
	
	public AbstractService() {
		
		initializeService();
	}

	protected void initializeService() {
		// Properties is serializable.  This code is called twice.
		// At the first time, when the Java application is run, it is called locally to generate
		// the SPL file.  At this time, the properties file should be accessible
		// and we read the data into a properties object.
		// At the second time, the Java object is being deserialized.  At this time, the properties
		// object is sent to the remote host as part of the seraizlied object.  We do not need
		// to read the file again and the properties attributes will still be available.
		if (properties == null)
			properties = readProperties();		
	}
	
	protected  Properties readProperties()
	{
		try {
			FileInputStream iStream = new FileInputStream("service.properties");
			Properties properties = new Properties();
			properties.load(iStream);
			iStream.close();
			
			return properties;
			
		} catch (FileNotFoundException e) {
			TRACE.error("Unable to find service.properties");
		} catch (IOException e) {
			TRACE.error("Unable to read service.properties");
		}
		return null;
	}
	
	public Properties getProperties() {
		return properties;
	}

	protected void addDependencies(Topology topology) {

		File dir = new File("opt/lib");
		
		// only include the required Jar files in classpath
		// exclude any IBM Streams jar files and topology jar files
		File[] jarFiles = dir.listFiles(new FilenameFilter() {
			@Override
			public boolean accept(File dir, String name) {
				return name.endsWith("jar") && (name.indexOf("streams.") == -1)
						&& (name.indexOf("com.ibm.streamsx.topology") == -1);
			}
		});

		if (jarFiles != null) {
			for (File file : jarFiles) {
				topology.addJarDependency(file.getAbsolutePath());
			}
		}
		
		File etcDir = new File("etc");
		if (etcDir.exists() && etcDir.isDirectory())
		{
			File[] etcFiles = etcDir.listFiles();
			for (int i = 0; i < etcFiles.length; i++) {
				topology.addFileDependency(etcFiles[i].getAbsolutePath(), "etc");
			}
		}
		
		topology.addClassDependency(AbstractService.class);
	}
	
	protected boolean isDebug()
	{
		String property = getProperties().getProperty(ICommonServicesConstants.KEY_DEBUG);
		if (property!=null && !property.isEmpty())
		{
			return Boolean.valueOf(property);
		}
		return false;
	}
	
	protected String getStreamsContext() {
		String property = getProperties().getProperty(ICommonServicesConstants.KEY_STREAMS_CONTEXT);
		if (property !=null && !property.isEmpty())
		{
			return property.toUpperCase();
		}
		return "DISTRIBUTED";
	}
	
	protected String getVmArgs()
	{
		return getProperties().getProperty(ICommonServicesConstants.KEY_VMARGS);
	}

	protected void submit(Topology topology) {
		try {
			Map<String, Object> subProperties = new HashMap<>();
	
			String vmArgs = getVmArgs();
	
			if (vmArgs != null && !vmArgs.isEmpty()) {
				// Add addition VM Arguments as specified in properties file
				subProperties.put(ContextProperties.VMARGS, vmArgs);
			}
	
			if (isDebug())
				subProperties.put(ContextProperties.TRACING_LEVEL, TraceLevel.DEBUG);
			
			String dataDir = (String)getProperties().get(ICommonServicesConstants.KEY_DATADIR);
			
			if (dataDir != null && !dataDir.isEmpty())
			{
				subProperties.put(JobProperties.DATA_DIRECTORY, dataDir);
			}
			
			addSubmssionTimeParams(subProperties);
	
			StreamsContextFactory.getStreamsContext(getStreamsContext()).submit(topology, subProperties);
		} catch (Exception e) {
			TRACE.error("Unable to submit topology", e);
		}
	}
	
	protected Map<String, Object> addSubmssionTimeParams(Map<String, Object> params) {
		return params;
	}
	
	protected String getStreamsInstall() {
		Map<String, String> envVars = System.getenv();
		return envVars.get("STREAMS_INSTALL");
	}

	protected void run() {
		
		Topology topo = createTopology();
		addDependencies(topo);
		submit(topo);
		
	}
	
	protected abstract Topology createTopology();
	

	
}

@@ -4370,16 +4370,208 @@
 ))%0A%09%09%09%7B%0A
+%09%09%09%09File file = new File(dataDir);%0A%09%09%09%09if (!file.isAbsolute())%0A%09%09%09%09%7B%0A%09%09%09%09%09String currentDirectory = System.getProperty(%22user.dir%22);%0A%09%09%09%09%09dataDir = currentDirectory + %22/%22 + dataDir;%0A%09%09%09%09%7D%0A%09%09%09%09%0A
 %09%09%09%09subP

Handle relative path for data dir