[jbossws-commits] JBossWS SVN: r1305 - in branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing: . mgmt

Author: heiko.braun(a)jboss.com Date: 2006-10-25 06:55:58 -0400 (Wed, 25 Oct 2006) New Revision: 1305 Modified: branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/EventingConstants.java branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManager.java branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManagerMBean.java Log: Added Stefano Maestri contribution Modified: branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/EventingConstants.java =================================================================== --- branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/EventingConstants.java 2006-10-25 10:55:06 UTC (rev 1304) +++ branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/EventingConstants.java 2006-10-25 10:55:58 UTC (rev 1305) @@ -79,8 +79,6 @@ // default config public final static String DISPATCHER_JNDI_NAME = "EventDispatcher"; - public final static long MAX_LEASE_TIME = 1000 * 60 * 10L; - public final static long DEFAULT_LEASE = 1000 * 60 * 5L; private static URI DELIVERY_PUSH_URI = null; private static URI DIALECT_XPATH_URI = null; Modified: branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManager.java =================================================================== --- branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManager.java 2006-10-25 10:55:06 UTC (rev 1304) +++ branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManager.java 2006-10-25 10:55:58 UTC (rev 1305) @@ -23,25 +23,6 @@ // $Id$ -import java.io.PrintWriter; -import java.io.StringWriter; -import java.net.URI; -import java.net.URISyntaxException; -import java.util.Date; -import java.util.List; -import java.util.ArrayList; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - -import javax.naming.InitialContext; -import javax.naming.NamingException; -import javax.management.*; - import org.jboss.logging.Logger; import org.jboss.util.naming.Util; import org.jboss.ws.WSException; @@ -49,10 +30,24 @@ import org.jboss.ws.eventing.deployment.EventingEndpointDI; import org.jboss.ws.eventing.element.EndpointReference; import org.jboss.ws.eventing.element.ReferenceParameters; +import org.jboss.ws.utils.ObjectNameFactory; import org.jboss.ws.utils.UUIDGenerator; -import org.jboss.ws.utils.ObjectNameFactory; import org.w3c.dom.Element; +import javax.management.MBeanServer; +import javax.management.MBeanServerFactory; +import javax.management.ObjectName; +import javax.naming.InitialContext; +import javax.naming.NamingException; +import java.io.PrintWriter; +import java.io.StringWriter; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.*; + /** * The SubscriptionManager maintains event sources and subscriptions.<br> * It is interfaced through the EventSourceEndpoint and SubscriptionManagerEndpoint SOAP endpoints. @@ -80,11 +75,12 @@ */ public class SubscriptionManager implements SubscriptionManagerMBean, EventDispatcher { - private static final Logger log = Logger.getLogger(SubscriptionManager.class); + private long MAX_LEASE_TIME = 1000 * 60 * 10L; + public long DEFAULT_LEASE = 1000 * 60 * 5L; public static final ObjectName OBJECT_NAME = ObjectNameFactory.create("jboss.ws:service=SubscriptionManager,module=eventing"); - + /** * Maps event source namespaces to event source instances. */ @@ -137,31 +133,31 @@ public void start() throws Exception { log.debug("Start subscription manager"); - + // setup thread pool threadPool = new ThreadPoolExecutor(5, 15, // core/max num threads - 5000, TimeUnit.MILLISECONDS, // 5 seconds keepalive - eventQueue); + 5000, TimeUnit.MILLISECONDS, // 5 seconds keepalive + eventQueue); // start the subscription watchdog watchDog = new WatchDog(subscriptionMapping); watchDog.startup(); } - public void stop() + public void stop() { log.debug("Stop subscription manager"); try { // remove event dispatcher Util.unbind(new InitialContext(), EventingConstants.DISPATCHER_JNDI_NAME); - + // stop thread pool threadPool.shutdown(); - + // stop the watchdog watchDog.shutdown(); - + for (URI eventSourceNS : eventSourceMapping.keySet()) { removeEventSource(eventSourceNS); @@ -171,7 +167,7 @@ { //log.warn("Cannot unbind event dispatcher: " + e.toString()); // Ignore - } + } } private static URI generateSubscriptionID() @@ -283,7 +279,7 @@ } else { - expires = new Date((System.currentTimeMillis() + EventingConstants.DEFAULT_LEASE)); + expires = new Date((System.currentTimeMillis() + DEFAULT_LEASE)); } // filter constraints @@ -324,8 +320,14 @@ private void assertLeaseConstraints(Date expireDate) throws SubscriptionError { long expires = expireDate.getTime() - System.currentTimeMillis(); - if (expires < 0 || EventingConstants.MAX_LEASE_TIME < expires) + if (expires <= 0 ) { throw new SubscriptionError(EventingConstants.CODE_INVALID_EXPIRATION_TIME, "The expiration time requested is invalid: " + expires + "ms"); + } + if (MAX_LEASE_TIME < expires) { + expireDate.setTime((System.currentTimeMillis() + MAX_LEASE_TIME)); + + } + } /** @@ -344,7 +346,7 @@ if (lease != null) assertLeaseConstraints(lease); - else lease = new Date((System.currentTimeMillis() + EventingConstants.DEFAULT_LEASE)); + else lease = new Date((System.currentTimeMillis() + DEFAULT_LEASE)); subscription.setExpires(lease); return lease; @@ -489,6 +491,11 @@ threadPool.setCorePoolSize(corePoolSize); } + public int getMaxPoolSize() + { + return threadPool.getMaximumPoolSize(); + } + public void setMaxPoolSize(int maxPoolSize) { threadPool.setMaximumPoolSize(maxPoolSize); @@ -499,6 +506,27 @@ threadPool.setKeepAliveTime(millies, TimeUnit.MILLISECONDS); } + public long getEventKeepAlive() { + return threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS); + } + + public void setDefaultLeaseTimeMins(int mins) { + this.DEFAULT_LEASE = mins * 60 * 1000L; + } + + public int getDefaultLeaseTimeMins() { + return (int) this.DEFAULT_LEASE/60/1000; + } + + public void setMaxLeaseTimeMins(int mins) { + this.MAX_LEASE_TIME = mins * 60 * 1000L; + } + + public int getMaxLeaseTimeMins() { + return (int) this.MAX_LEASE_TIME/60/1000; + } + + /** * The watchdog maintains subscription expirations. */ Modified: branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManagerMBean.java =================================================================== --- branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManagerMBean.java 2006-10-25 10:55:06 UTC (rev 1304) +++ branches/jbossws-1.0/src/main/java/org/jboss/ws/eventing/mgmt/SubscriptionManagerMBean.java 2006-10-25 10:55:58 UTC (rev 1305) @@ -32,26 +32,44 @@ /** * @author Heiko Braun, <heiko(a)openj.net&gt; + * @author Stefano Maestri * @since 12-Dec-2005 */ public interface SubscriptionManagerMBean { public static final String BEAN_NAME = "SubscriptionManager"; + // ---------------------------------------------------------------------------- + // Read/Write attributes + /** * Returns the core number of threads. * * @return long */ - int getCorePoolSize(); + public int getCorePoolSize(); + public void setCorePoolSize(int corePoolSize); /** * Returns the maximum allowed number of threads. * * @return int */ - int getMaximumPoolSize(); + public int getMaxPoolSize(); + public void setMaxPoolSize(int maxPoolSize); + public long getEventKeepAlive(); + public void setEventKeepAlive(long millies); + + public void setMaxLeaseTimeMins(int mins); + public int getMaxLeaseTimeMins(); + + public void setDefaultLeaseTimeMins(int mins); + public int getDefaultLeaseTimeMins(); + + // ---------------------------------------------------------------------------- + // Read only attributes + /** * Returns the largest number of threads that have ever simultaneously been in the pool. * @@ -73,12 +91,7 @@ */ long getCompletedTaskCount(); - public void setCorePoolSize(int corePoolSize); - - public void setMaxPoolSize(int maxPoolSize); - - public void setEventKeepAlive(long millies); - + // ---------------------------------------------------------------------------- // subscription endpointReference business SubscriptionTicket subscribe(URI eventSourceNS, EndpointReference notifyTo, EndpointReference endTo, Date expires, Filter filter) throws SubscriptionError;