common/src/main/java/org/apache/abdera2/common/protocol/servlet/async/AbderaAsyncService.java - abdera - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  The ASF licenses this file to You
  * under the Apache License, Version 2.0 (the "License"); you may not
  * use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.  For additional information regarding
  * copyright in this work, please see the NOTICE file in the top level
  * directory of this distribution.
  */
 package org.apache.abdera2.common.protocol.servlet.async;

 import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Map;

 import javax.servlet.ServletContext;
 import javax.servlet.ServletContextEvent;
 import javax.servlet.ServletContextListener;
 import javax.servlet.annotation.WebListener;

 import org.apache.abdera2.common.protocol.Provider;
 import org.apache.abdera2.common.protocol.ServiceManager;
 import org.apache.abdera2.common.pusher.ChannelManager;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 @WebListener
 public class AbderaAsyncService
   implements ServletContextListener, Runnable {

     private static final int DEFAULT_WORKER_THREADS = 10;
     public static final String PROPERTY_WORKER_THREADS = "AbderaWorkerThreadCount";
     public static final String PROPERTY_ATOMPUB_SERVICE = "AbderaAtompubService";
     public static final String PROPERTY_CHANNEL_SERVICE = "AbderaChannelService";

     public static final String RUNNER = "AbderaRunner";
     public static final String SERVICEMANAGER = "AbderaServiceManager";
     public static final String PROVIDER = "AbderaProvider";
     public static final String QUEUE = "AbderaProcessorQueue";
     public static final String CM = "AbderaChannelManager";

     private final static Log log = LogFactory.getLog(AbderaAsyncService.class);

     private ServletContext context;
     private TaskExecutor exec;
     private ProcessorQueue queue;
     private ChannelManager cm;
     private boolean deployAtompubService;
     private boolean deployChannelService;

     public AbderaAsyncService() {
       log.debug("Abdera Async Service Created");
     }

     protected Map<String, String> getProperties(ServletContext context) {
       Map<String, String> properties = new HashMap<String, String>();
       Enumeration<String> e = context.getInitParameterNames();
       while (e.hasMoreElements()) {
           String key = e.nextElement();
           String val = context.getInitParameter(key);
           properties.put(key, val);
       }
       return properties;
     }

     private int worker_threads(Map<String,String> properties) {
       int c = DEFAULT_WORKER_THREADS;
       if (properties.containsKey(PROPERTY_WORKER_THREADS)) {
         String val = properties.get(PROPERTY_WORKER_THREADS);
         c = Math.max(1,Integer.parseInt(val));
       }
       return c;
     }

     private boolean getBooleanProperty(Map<String, String> properties, String name, boolean def) {
       boolean answer = def;
       if (properties.containsKey(name)) {
         String val = properties.get(name);
         answer = "TRUE".equalsIgnoreCase(val) ||
                  "1".equals(val) ||
                  "YES".equalsIgnoreCase(val);
       }
       return answer;
     }

     public boolean isDeployAtompubService(Map<String,String> properties) {
       return getBooleanProperty(properties,PROPERTY_ATOMPUB_SERVICE,false);
     }

     public boolean isDeployChannelService(Map<String,String> properties) {
       return getBooleanProperty(properties,PROPERTY_CHANNEL_SERVICE,false);
     }

     @SuppressWarnings("unchecked")
     protected ServiceManager<Provider> createServiceManager(ServletContext context) {
       String prop = context.getInitParameter(ServiceManager.class.getName());
       return prop != null ?
         ServiceManager.Factory.getInstance(prop) :
         ServiceManager.Factory.getInstance();
     }


     public void contextInitialized(ServletContextEvent event) {
       this.context = event.getServletContext();
       Map<String,String> properties = getProperties(context);
       this.deployAtompubService = isDeployAtompubService(properties);
       this.deployChannelService = isDeployChannelService(properties);
       ServiceManager<Provider> manager =
         createServiceManager(context);

       if (manager == null) {
           log.error("Service Manager is null. Application can not function correctly");
           throw new IllegalStateException("Service Manager is null");
       }

       if (deployAtompubService) {
         log.debug("Initializing Abdera Atompub Service...");
         queue = manager.newProcessorQueue(properties);
         exec = manager.newTaskExecutor(properties);
         Provider provider = manager.newProvider(properties);
         Processor processor = queue != null ? queue.getProcessor() : null;

         log.debug(String.format("Queue:           %s",queue));
         log.debug(String.format("Processor:       %s",processor));
         log.debug(String.format("Executor:        %s",exec));
         log.debug(String.format("Service Manager: %s",manager));
         log.debug(String.format("Provider:        %s",provider));

         if (processor != null)
           context.setAttribute(Processor.NAME, processor);
         else {
           log.error("Queue Processor is null. Application can not function correctly");
           throw new IllegalStateException("Queue Processor is null");
         }
         if (exec != null)
           context.setAttribute(RUNNER, exec);
         else {
           log.error("Task Executor is null. Application can not function correctly");
           throw new IllegalStateException("Task Executor is null");
         }
         if (provider != null)
           context.setAttribute(PROVIDER, provider);
         else {
           log.error("Provider is null. Application can not function correctly");
           throw new IllegalStateException("Provider is null");
         }
         if (queue != null)
           context.setAttribute(QUEUE, queue);
         else {
           log.error("Queue is null. Application can not function correctly");
           throw new IllegalStateException("Queue is null");
         }
         context.setAttribute(SERVICEMANAGER, manager);

         int ct = worker_threads(properties);
         log.debug(String.format("Launching watcher threads [%d]",ct));

         for (int c = 0; c < ct; c++)
           exec.startWorker(this);

         log.debug("Abdera Atompub Service is ready...");
       }

       if (deployChannelService) {
         log.debug("Initializing Abdera Channel Service");
         cm = manager.newChannelManager(properties);
         log.debug(String.format("Channel Manager: %s", cm));
         if (cm != null) {
           context.setAttribute(CM, cm);
           log.debug("Abdera Channel Service is ready...");
         } else log.debug("Abdera Channel Service could not be started");
       }
     }

     public void contextDestroyed(ServletContextEvent event) {
       ServletContext context = event.getServletContext();
       if (deployAtompubService) {
         log.debug("Shutting down the Abdera Service...");
         if (exec != null)
           exec.shutdown();
         // if there are remaining outstanding requests after
         // shutdown we need to deal with them
         if (queue != null)
           queue.cancelRemaining();

         context.removeAttribute(Processor.NAME);
         context.removeAttribute(RUNNER);
         context.removeAttribute(SERVICEMANAGER);
         context.removeAttribute(PROVIDER);
         context.removeAttribute(QUEUE);
       }
       if (deployChannelService) {
         if (cm != null)
           cm.shutdown();
         context.removeAttribute(CM);
       }
     }

     public void run() {
       TaskExecutor exec =
         (TaskExecutor) context.getAttribute(RUNNER);
       ProcessorQueue processor =
         (ProcessorQueue) context.getAttribute(QUEUE);
       while(exec.isRunning()) {
         if (processor.hasNext()) {
           final AbderaTask task = processor.next();
           log.debug(String.format("Processing New AbderaTask (%s)...",task.getId()));
           exec.execute(new Runnable() {
             public void run() {
               try {
                 task.invoke();
               } catch (Throwable t) {
                 log.error(String.format("Error invoking AbderaTask (%s)",task.getId()),t);
               }
               log.debug(String.format("AbderaTask (%s) is complete",task.getId()));
             }
           });
         }
       }
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. The ASF licenses this file to You
	* under the Apache License, Version 2.0 (the "License"); you may not
	* use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License. For additional information regarding
	* copyright in this work, please see the NOTICE file in the top level
	* directory of this distribution.
	*/
	package org.apache.abdera2.common.protocol.servlet.async;

	import java.util.Enumeration;
	import java.util.HashMap;
	import java.util.Map;

	import javax.servlet.ServletContext;
	import javax.servlet.ServletContextEvent;
	import javax.servlet.ServletContextListener;
	import javax.servlet.annotation.WebListener;

	import org.apache.abdera2.common.protocol.Provider;
	import org.apache.abdera2.common.protocol.ServiceManager;
	import org.apache.abdera2.common.pusher.ChannelManager;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	@WebListener
	public class AbderaAsyncService
	implements ServletContextListener, Runnable {

	private static final int DEFAULT_WORKER_THREADS = 10;
	public static final String PROPERTY_WORKER_THREADS = "AbderaWorkerThreadCount";
	public static final String PROPERTY_ATOMPUB_SERVICE = "AbderaAtompubService";
	public static final String PROPERTY_CHANNEL_SERVICE = "AbderaChannelService";

	public static final String RUNNER = "AbderaRunner";
	public static final String SERVICEMANAGER = "AbderaServiceManager";
	public static final String PROVIDER = "AbderaProvider";
	public static final String QUEUE = "AbderaProcessorQueue";
	public static final String CM = "AbderaChannelManager";

	private final static Log log = LogFactory.getLog(AbderaAsyncService.class);

	private ServletContext context;
	private TaskExecutor exec;
	private ProcessorQueue queue;
	private ChannelManager cm;
	private boolean deployAtompubService;
	private boolean deployChannelService;

	public AbderaAsyncService() {
	log.debug("Abdera Async Service Created");
	}

	protected Map<String, String> getProperties(ServletContext context) {
	Map<String, String> properties = new HashMap<String, String>();
	Enumeration<String> e = context.getInitParameterNames();
	while (e.hasMoreElements()) {
	String key = e.nextElement();
	String val = context.getInitParameter(key);
	properties.put(key, val);
	}
	return properties;
	}

	private int worker_threads(Map<String,String> properties) {
	int c = DEFAULT_WORKER_THREADS;
	if (properties.containsKey(PROPERTY_WORKER_THREADS)) {
	String val = properties.get(PROPERTY_WORKER_THREADS);
	c = Math.max(1,Integer.parseInt(val));
	}
	return c;
	}

	private boolean getBooleanProperty(Map<String, String> properties, String name, boolean def) {
	boolean answer = def;
	if (properties.containsKey(name)) {
	String val = properties.get(name);
	answer = "TRUE".equalsIgnoreCase(val) \|\|
	"1".equals(val) \|\|
	"YES".equalsIgnoreCase(val);
	}
	return answer;
	}

	public boolean isDeployAtompubService(Map<String,String> properties) {
	return getBooleanProperty(properties,PROPERTY_ATOMPUB_SERVICE,false);
	}

	public boolean isDeployChannelService(Map<String,String> properties) {
	return getBooleanProperty(properties,PROPERTY_CHANNEL_SERVICE,false);
	}

	@SuppressWarnings("unchecked")
	protected ServiceManager<Provider> createServiceManager(ServletContext context) {
	String prop = context.getInitParameter(ServiceManager.class.getName());
	return prop != null ?
	ServiceManager.Factory.getInstance(prop) :
	ServiceManager.Factory.getInstance();
	}


	public void contextInitialized(ServletContextEvent event) {
	this.context = event.getServletContext();
	Map<String,String> properties = getProperties(context);
	this.deployAtompubService = isDeployAtompubService(properties);
	this.deployChannelService = isDeployChannelService(properties);
	ServiceManager<Provider> manager =
	createServiceManager(context);

	if (manager == null) {
	log.error("Service Manager is null. Application can not function correctly");
	throw new IllegalStateException("Service Manager is null");
	}

	if (deployAtompubService) {
	log.debug("Initializing Abdera Atompub Service...");
	queue = manager.newProcessorQueue(properties);
	exec = manager.newTaskExecutor(properties);
	Provider provider = manager.newProvider(properties);
	Processor processor = queue != null ? queue.getProcessor() : null;

	log.debug(String.format("Queue: %s",queue));
	log.debug(String.format("Processor: %s",processor));
	log.debug(String.format("Executor: %s",exec));
	log.debug(String.format("Service Manager: %s",manager));
	log.debug(String.format("Provider: %s",provider));

	if (processor != null)
	context.setAttribute(Processor.NAME, processor);
	else {
	log.error("Queue Processor is null. Application can not function correctly");
	throw new IllegalStateException("Queue Processor is null");
	}
	if (exec != null)
	context.setAttribute(RUNNER, exec);
	else {
	log.error("Task Executor is null. Application can not function correctly");
	throw new IllegalStateException("Task Executor is null");
	}
	if (provider != null)
	context.setAttribute(PROVIDER, provider);
	else {
	log.error("Provider is null. Application can not function correctly");
	throw new IllegalStateException("Provider is null");
	}
	if (queue != null)
	context.setAttribute(QUEUE, queue);
	else {
	log.error("Queue is null. Application can not function correctly");
	throw new IllegalStateException("Queue is null");
	}
	context.setAttribute(SERVICEMANAGER, manager);

	int ct = worker_threads(properties);
	log.debug(String.format("Launching watcher threads [%d]",ct));

	for (int c = 0; c < ct; c++)
	exec.startWorker(this);

	log.debug("Abdera Atompub Service is ready...");
	}

	if (deployChannelService) {
	log.debug("Initializing Abdera Channel Service");
	cm = manager.newChannelManager(properties);
	log.debug(String.format("Channel Manager: %s", cm));
	if (cm != null) {
	context.setAttribute(CM, cm);
	log.debug("Abdera Channel Service is ready...");
	} else log.debug("Abdera Channel Service could not be started");
	}
	}

	public void contextDestroyed(ServletContextEvent event) {
	ServletContext context = event.getServletContext();
	if (deployAtompubService) {
	log.debug("Shutting down the Abdera Service...");
	if (exec != null)
	exec.shutdown();
	// if there are remaining outstanding requests after
	// shutdown we need to deal with them
	if (queue != null)
	queue.cancelRemaining();

	context.removeAttribute(Processor.NAME);
	context.removeAttribute(RUNNER);
	context.removeAttribute(SERVICEMANAGER);
	context.removeAttribute(PROVIDER);
	context.removeAttribute(QUEUE);
	}
	if (deployChannelService) {
	if (cm != null)
	cm.shutdown();
	context.removeAttribute(CM);
	}
	}

	public void run() {
	TaskExecutor exec =
	(TaskExecutor) context.getAttribute(RUNNER);
	ProcessorQueue processor =
	(ProcessorQueue) context.getAttribute(QUEUE);
	while(exec.isRunning()) {
	if (processor.hasNext()) {
	final AbderaTask task = processor.next();
	log.debug(String.format("Processing New AbderaTask (%s)...",task.getId()));
	exec.execute(new Runnable() {
	public void run() {
	try {
	task.invoke();
	} catch (Throwable t) {
	log.error(String.format("Error invoking AbderaTask (%s)",task.getId()),t);
	}
	log.debug(String.format("AbderaTask (%s) is complete",task.getId()));
	}
	});
	}
	}
	}

	}