CommEngine/src/main/java/altk/comm/engine/CommEngine.java

package altk.comm.engine;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.servlet.ServletContext;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import altk.comm.engine.Broadcast.BroadcastState;

@SuppressWarnings("serial")
public abstract class CommEngine extends HttpServlet
{    
    public static final String SERVICE_THREADPOOL_SIZE_KEY = "service_threadpool_size";

	static final String REQUEST_TOP_ELEMENT_NAME_DEFAULT = "Request";

    private static final int SCHEDULER_THREAD_POOL_SIZE = 1;
    
    private static final long DEAD_BROADCAST_VIEWING_PERIOD_DEFAULT = 60;

    private static final int SERVICE_THREADPOOL_SIZE_DEFAULT = 1;

    private static final int POSTBACK_THREADPOOL_SIZE_DEFAULT = 2;
    private static final int POSTBACK_MAX_QUEUE_SIZE_DEFAULT = 10000;
    private static final int POSTBACK_MAX_BATCH_SIZE_DEFAULT = 100;
    /**
     * Maps a broadcastId to a broadcast.
     */
    private Map<String, Broadcast> broadcasts;

    protected boolean notInService;
    
    protected Properties config;
    
    protected final String engineName; // e.g. "broadcast_sms", "broadcast_voice"

    private long startupTimestamp;

    // Sequencing naming of broadcast that fails to yield its broadcastId
    private int unknownBroadcastIdNdx = 1;

    /**
     * Used to communicate media-specific platform resources to broadcasts
     */
    protected EngineResources resources;
    
    protected static Logger myLogger;
    
    private ScheduledExecutorService scheduler;

    private long deadBroadcastViewingMinutes;

    private int completedJobCount = 0;

    private int serviceThreadPoolSize;

    private int postbackMaxQueueSize;

    private int postbackSenderPoolSize;

    private int postbackMaxBatchSize;

	protected String runtimeDirPath;
	protected String confDirPath;

    abstract protected Broadcast mkBroadcast();
    
    public CommEngine(String engineName)
    {
        this.engineName = engineName;
        broadcasts = new HashMap<String, Broadcast>();
        startupTimestamp = System.currentTimeMillis();
    }
    
    /**
     * Relocates a filepath relative to runtime directory if filepath is not absolute.
     * @param filepath
     * @return
     */
    public String relocateToRuntimeDir(String filepath)
    {
		if (filepath.startsWith("/")) return filepath; // no change to absolute path
	
		String relocated = filepath;
		
		// The next 2 lines take care of pre-git era meaning convention of filepath in properties.
		// Then, the runtime is relative to the current working directory of tomcat.
		// Now they are relative to the runtimDirPath obtained from the tomcat tomcat context.
		// These 2 lines an be deleted when all CommEngines in production are in the git era.
		String unwanted_prefix = engineName + "/";
		if (filepath.startsWith(unwanted_prefix)) relocated = filepath.substring(unwanted_prefix.length());
		
		relocated = runtimeDirPath + "/" + relocated;
		return relocated;
    }
    
    /**
     * Invoked by servlet container during initialization of servlet.
     */
    public final void init()
    {
        // check init parameters
        ServletContext servletContext = getServletContext();
        
        confDirPath = servletContext.getInitParameter(getConfDirContextName());
        System.out.println("Config directory is configured to be '" + confDirPath + "'.  Make sure it and its content are readable by user 'tomcat'");
        runtimeDirPath = servletContext.getInitParameter(getRunTimeDirContextName());
        System.out.println("Runtime directory is configured to be '" + runtimeDirPath + "'.  Make sure it and its content are readable by user 'tomcat'");
        
        File propertiesFile = new File(confDirPath + "/properties");
      
        // Configure log4j using log4j.properties file, \
        // sibling to the engine properties file.
        // This change is backward compatible with placing the log4j.properties file in
        // the class path.
        String log4j_properties = confDirPath + "/log4j.properties";
        // Relocate file property offetting it by the runtimeDirPath
        try
        {
        	Properties prop = new Properties();
        	prop.load(new FileInputStream(log4j_properties));
        	Enumeration<Object> e = prop.keys();
        	while (e.hasMoreElements())
        	{
        		String key = (String)e.nextElement();
        		if (key.toLowerCase().endsWith(".file"))
        		{
        			String filepath = prop.getProperty(key);
        			String relocate = relocateToRuntimeDir(filepath);
					prop.setProperty(key, relocate);
        			System.out.println(key + "=" + relocate);
        		}
        	}
            PropertyConfigurator.configure(prop);
        }
        catch (Exception e)
        {
        	System.out.println("Failed to configure log4: " + e);
            // Do nothing, assuming the exception is FileNotFoundException.
            // Remaining log4j initialization will look for log4j.properties
            // file in the class path.
        }
        // This activates Logger class instantiation.  At this point, if lo4j
        // is not yet configured,
        // it will look for the log4j.properties file in the class path.
        myLogger = Logger.getLogger(CommEngine.class);
        
        myLogger.info("init() invoked");
//        File propertiesFile = new File(
//        		getClass().getClassLoader().getResource("properties").getFile()
//        	);

        CommonLogger.startup.info("Using lo4j properites file " + log4j_properties);
        CommonLogger.startup.info("Using configuration file " + propertiesFile.getAbsolutePath());
        config = new Properties();
        try
        {
            config.load(new FileInputStream(propertiesFile));
        }
        catch (FileNotFoundException e)
        {
            CommonLogger.alarm.fatal("Properties file " + propertiesFile.getAbsolutePath() + " not found -- abort");
            notInService = true;
            return;
        }
        catch (IOException e)
        {
            CommonLogger.alarm.fatal("Problem in reading properties file " + propertiesFile.getAbsolutePath() + ": " + e.getMessage());
            notInService = true;
            return;
        }
        
        // Set up periodic purge of stale broadcasts, based on deadBroadcastViewingMinutes
        String periodStr = config.getProperty("dead_broadcast_viewing_period",
                new Long(DEAD_BROADCAST_VIEWING_PERIOD_DEFAULT).toString());
        deadBroadcastViewingMinutes = Long.parseLong(periodStr);
        CommonLogger.startup.info(String.format("Dead broadcast viewing period: %d minutes", deadBroadcastViewingMinutes));

        String str = config.getProperty(SERVICE_THREADPOOL_SIZE_KEY,
                new Integer(SERVICE_THREADPOOL_SIZE_DEFAULT).toString());
        serviceThreadPoolSize = Integer.parseInt(str);
        CommonLogger.startup.info(String.format("service thread pool size: %d", serviceThreadPoolSize));

        String string = config.getProperty("postback_max_queue_size",
                new Integer(POSTBACK_MAX_QUEUE_SIZE_DEFAULT).toString());
        postbackMaxQueueSize  = Integer.parseInt(string);
        CommonLogger.activity.info("Postback max queue size = " + postbackMaxQueueSize);

        string = config.getProperty("postback_threadpool_size",
                new Integer(POSTBACK_THREADPOOL_SIZE_DEFAULT).toString());
        postbackSenderPoolSize = Integer.parseInt(string);
        CommonLogger.activity.info("Postback threadpool size = " + postbackSenderPoolSize);
        
        string = config.getProperty("postback_max_batch_size",
                new Integer(POSTBACK_MAX_BATCH_SIZE_DEFAULT).toString());
        postbackMaxBatchSize = Integer.parseInt(string);
        CommonLogger.activity.info("Postback max batch size = " + postbackMaxBatchSize);
        
        scheduler = Executors.newScheduledThreadPool(SCHEDULER_THREAD_POOL_SIZE);
        scheduler.scheduleAtFixedRate(new Runnable() { public void run() { purgeStaleBroadcasts();}},
                deadBroadcastViewingMinutes, deadBroadcastViewingMinutes, TimeUnit.MINUTES);

        initChild();
    }

    protected void purgeStaleBroadcasts()
    {
        long now = System.currentTimeMillis();
        synchronized (broadcasts)
        {
            for (String id : broadcasts.keySet())
            {
                if (now - broadcasts.get(id).changeStateTime > deadBroadcastViewingMinutes * 60 * 1000)
                {
                    Broadcast broadcast = broadcasts.get(id);
                    completedJobCount += broadcast.getCompletedJobCount();
                    broadcasts.remove(id);
                }
            }
        }
    }
    
    /**
     * 
     * @return name of parameter in Tomcat context file, specifying properties file
     * for this SMSEngine.
     */
    protected abstract String getConfDirContextName();
    
    protected abstract String getRunTimeDirContextName();

    @Override
    protected void doPost(HttpServletRequest request, HttpServletResponse response)
    {
        Broadcast broadcast = mkBroadcast();
        broadcast.doPost(request, response, this);
    }

    /**
     * Functions covered are 
     *   get=status
     *   get=cancel_broadcast&broadcast_id=XXX
     */
    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response)
    {
        PrintWriter out;
        try
        {
            out = response.getWriter();
        }
        catch (IOException e)
        {
            CommonLogger.alarm.error("Cannot write a reply: " + e);
            return;
        }
        String get = (String)request.getParameter("get");
        if (get == null)
        {
            // Return http status BAD REQUEST
            int httpStatus = HttpServletResponse.SC_BAD_REQUEST;
            try
            {
                response.sendError(httpStatus);
            }
            catch (IOException e)
            {
                myLogger.warn("Unnable to return HTTP error code " + httpStatus);
            }
            return;
        }
        if (get.equalsIgnoreCase("status"))
        {
            getStatus(request, out);
        }
        else if (get.equalsIgnoreCase("cancel_broadcast"))
        {
            cancelBroadcast(request, out);
        }
        else if (get.equalsIgnoreCase("pause_broadcast"))
        {
            pauseBroadcast(request, out);
        }
        else if (get.equalsIgnoreCase("resume_broadcast"))
        {
            resumeBroadcast(request, out);
        }
        else
        {
            out.write(get + " not supported");
        }
        out.close();
    }

    private void cancelBroadcast(HttpServletRequest request, PrintWriter out)
    {
        // Get broadcastId from request
        String broadcastId = getBroadcastId(request);
        Broadcast broadcast = broadcasts.get(broadcastId);
        if (broadcast == null)
        {
            out.format("Broadcast %s does not exist", broadcastId);
            return;
        }
        broadcast.cancel(out);
    }
    
    protected void pauseBroadcast(HttpServletRequest request, PrintWriter out)
    {
	    // Get broadcastId from request
	    String broadcastId = getBroadcastId(request);
	    Broadcast broadcast = broadcasts.get(broadcastId);
	    if (broadcast == null)
	    {
	        out.format("Broadcast %s does not exist", broadcastId);
	        return;
	    }
	    broadcast.pause(out);
    }


    protected void resumeBroadcast(HttpServletRequest request, PrintWriter out)
    {
        // Get broadcastId from request
        String broadcastId = getBroadcastId(request);
        Broadcast broadcast = broadcasts.get(broadcastId);
        if (broadcast == null)
        {
            out.format("Broadcast %s does not exist", broadcastId);
            return;
        }
        broadcast.resume(out);
    }

    /**
     * <CallEngine_status>
     *   status of each broadcast
     *   <calls><total>ttt</total><connected>nnn</connected>
     * </CallEngine_status>
     */
    private void getStatus(HttpServletRequest request, PrintWriter out)
    {
        String broadcastId = request.getParameter("broadcast_id");
        if (broadcastId != null)
        {
            broadcastId = broadcastId.trim();
            if (broadcastId.length() == 0)
            {
                out.write("broadcast_id request parameter cannot be empty");
                return;
            }
            Broadcast broadcast = broadcasts.get(broadcastId);
            if (broadcast == null)
            {
                out.write("<error>No such broadcast</error>");
            }
            else
            {
                out.write(broadcast.mkStatusReport());
            }
            return;
        }
        else
        {
            String tag = engineName + "_status";
            out.write("<" + tag + ">\r\n");
            
            out.write("<startup_time>" + startupTimestamp
                    + "</startup_time>\r\n");
            
            // First get a copy of broadcasts, to avoid mutex deadlock.
            Vector<Broadcast> broadcastList = new Vector<Broadcast>();
            synchronized(broadcasts)
            {
                for (String key : broadcasts.keySet())
                {
                    broadcastList.add(broadcasts.get(key));
                }
            }
            // We have released the lock.
            // Then append status of each broadcast to outBuf.
            for (Broadcast broadcast : broadcastList)
            {
                out.write(broadcast.mkStatusReport());
            }
            out.write("<job_summary completed='" + getCompletedJobCount() + "' pending='" + getPendingJobCount() + "' active='" + getActiveJobCount() + "'/>");
            out.write("</" + tag + ">");
        }
    }

    public int getPendingJobCount()
    {
        int readyCount = 0;
        synchronized(broadcasts)
        {
            for (Broadcast broadcast : broadcasts.values())
            {
                readyCount += broadcast.getPendingJobCount();
            }
        }
        return readyCount;
        
    }

    public int getActiveJobCount()
    {
        int activeCount = 0;
        synchronized(broadcasts)
        {
            for (Broadcast broadcast : broadcasts.values())
            {
                activeCount += broadcast.getActiveJobCount();
            }
        }
        return activeCount;
    }

    public int getCompletedJobCount()
    {
        int additionalCompletedJobCount = 0;
        synchronized(broadcasts)
        {
            for (Broadcast broadcast : broadcasts.values())
            {
                additionalCompletedJobCount += broadcast.getCompletedJobCount();
            }
        }
        return completedJobCount  + additionalCompletedJobCount;
    }

    public void removeBroadcast(String broadcastId)
    {
        CommonLogger.activity.info("Removing broadcast " + broadcastId);
        synchronized(broadcasts)
        {
            broadcasts.remove(broadcastId);
        }
    }

	public boolean notInService()
    {
        return notInService;
    }

    /**
     * Decode http GET request for broadcast_id value
     * @param request
     * @return broadcast_id
     */
    private String getBroadcastId(HttpServletRequest request)
    {
        return request.getParameter("broadcast_id");
    }
    
    /**
     * Invoked by servlet container when servlet is destroyed.
     */
    public final void destroy()
    {
        System.out.println("Destroying " + engineName);
        
        // Shutdown threads that periodically purge stale broadcasts.
        scheduler.shutdownNow();
        
        for (Broadcast broadcast : broadcasts.values())
        {
        	broadcast.terminate(BroadcastState.ABORTED, "Platform termination");
        }
        
        destroyChild();
        super.destroy();
    }

    /**
     * Indirectly invoked by servlet container during servlet initialization.
     */
    abstract protected void initChild();
    
    /**
     * Indirectly invoked by servlet container during destruction of servlet.
     */
    abstract protected void destroyChild();

    public EngineResources getResources()
    {
        return resources;
    }

    public void addBroadcast(String broadcastId, Broadcast broadcast)
    {
        if (broadcastId == null) broadcastId = "Unknown" + unknownBroadcastIdNdx++;
        broadcasts.put(broadcastId,  broadcast);
        
    }

    /**
     * If broadcast has no id, one will be created for it.
     * @param broadcast
     */
    public void installBroadcast(Broadcast broadcast)
    {
        String broadcastId = broadcast.getBroadcastId();
        if (broadcastId == null) broadcastId = "Unknown" + unknownBroadcastIdNdx++;
        broadcasts.put(broadcastId,  broadcast);
    }

    @Deprecated
    public int getServiceThreadPoolSize()
    {
        return serviceThreadPoolSize;
    }

	public int getPostbackMaxQueueSize() {
		return postbackMaxQueueSize;
	}

	public int getPostbackSenderPoolSize() {
		return postbackSenderPoolSize;
	}

	public int getPostbackMaxBatchSize() {
		return postbackMaxBatchSize;
	}
}