Author: nbelaevski Date: 2011-02-10 09:53:59 -0500 (Thu, 10 Feb 2011) New Revision: 21608 Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/AtmospherePushHandler.java trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionManagerImpl.java trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionQueue.java trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/MessagingContext.java trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/PushContextImpl.java Log: https://issues.jboss.org/browse/RF-10487 Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/AtmospherePushHandler.java =================================================================== --- trunk/core/impl/src/main/java/org/richfaces/application/push/impl/AtmospherePushHandler.java 2011-02-10 14:52:12 UTC (rev 21607) +++ trunk/core/impl/src/main/java/org/richfaces/application/push/impl/AtmospherePushHandler.java 2011-02-10 14:53:59 UTC (rev 21608) @@ -25,7 +25,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; -import java.util.concurrent.atomic.AtomicInteger; import javax.servlet.ServletConfig; import javax.servlet.http.HttpServletRequest; @@ -37,25 +36,25 @@ import org.richfaces.application.push.Request; import org.richfaces.application.push.Session; import org.richfaces.application.push.SessionManager; +import org.richfaces.log.Logger; +import org.richfaces.log.RichfacesLogger; +import com.google.common.util.concurrent.ThreadFactoryBuilder; + /** * @author Nick Belaevski * */ public abstract class AtmospherePushHandler implements AtmosphereHandler<HttpServletRequest, HttpServletResponse> { - private static final ThreadFactory DAEMON_THREADS_FACTORY = new ThreadFactory() { + private static final Logger LOGGER = RichfacesLogger.APPLICATION.getLogger(); + + private static final ThreadFactory BROADCASTER_THREADS_FACTORY = new ThreadFactoryBuilder(). + setDaemon(true).setNameFormat("rf-push-worker-thread-%1$s").build(); - private final AtomicInteger threadsCounter = new AtomicInteger(); + private static final ThreadFactory SESSION_MANAGER_THREADS_FACTORY = new ThreadFactoryBuilder(). + setDaemon(true).setNameFormat("rf-push-session-manager-thread-%1$s").build(); - public Thread newThread(Runnable r) { - Thread t = new Thread(r, "rf-push-worker-thread-" + threadsCounter.getAndIncrement()); - t.setDaemon(true); - - return t; - } - }; - private static final String PUSH_SESSION_ID_PARAM = "pushSessionId"; private SessionManager sessionManager; @@ -65,8 +64,8 @@ public AtmospherePushHandler() { super(); - sessionManager = new SessionManagerImpl(DAEMON_THREADS_FACTORY); - worker = Executors.newCachedThreadPool(DAEMON_THREADS_FACTORY); + sessionManager = new SessionManagerImpl(SESSION_MANAGER_THREADS_FACTORY); + worker = Executors.newCachedThreadPool(BROADCASTER_THREADS_FACTORY); } public SessionManager getSessionManager() { @@ -119,7 +118,17 @@ } public void destroy() throws Exception { - sessionManager.destroy(); + try { + worker.shutdown(); + } catch (Exception e) { + LOGGER.error(e.getMessage(), e); + } + + try { + sessionManager.destroy(); + } catch (Exception e) { + LOGGER.error(e.getMessage(), e); + } } } Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionManagerImpl.java =================================================================== --- trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionManagerImpl.java 2011-02-10 14:52:12 UTC (rev 21607) +++ trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionManagerImpl.java 2011-02-10 14:53:59 UTC (rev 21608) @@ -21,8 +21,9 @@ */ package org.richfaces.application.push.impl; -import java.util.Iterator; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import org.richfaces.application.push.Session; @@ -47,28 +48,34 @@ } private final class SessionsExpirationRunnable implements Runnable { + public void run() { - while (true) { - try { - Session session = sessionQueue.take(); - sessionMap.remove(session.getId()); - if (session instanceof DestroyableSession) { - ((DestroyableSession) session).destroy(); - } - } catch (InterruptedException e) { - LOGGER.error(e.getMessage(), e); + try { + Session session = sessionQueue.take(); + + if (session instanceof DestroyableSession) { + ((DestroyableSession) session).destroy(); } + + sessionMap.remove(session.getId()); + + executorService.submit(this); + } catch (InterruptedException e) { + LOGGER.debug(e.getMessage(), e); } + } - } } private ConcurrentMap<String, Session> sessionMap = new MapMaker().makeMap(); private SessionQueue sessionQueue = new SessionQueue(); - + + private ExecutorService executorService; + public SessionManagerImpl(ThreadFactory threadFactory) { - threadFactory.newThread(new SessionsExpirationRunnable()).start(); + executorService = Executors.newSingleThreadExecutor(threadFactory); + executorService.submit(new SessionsExpirationRunnable()); } public Session getPushSession(String id) { @@ -76,16 +83,12 @@ } public void destroy() { - //TODO notify all session - sessionQueue.clear(); + executorService.shutdown(); + sessionQueue.shutdown(); - while (!sessionMap.isEmpty()) { - for (Iterator<Session> sessionsItr = sessionMap.values().iterator(); sessionsItr.hasNext(); ) { - Session session = sessionsItr.next(); - - if (session instanceof DestroyableSession) { - ((DestroyableSession) session).destroy(); - } + for (Session session: sessionMap.values()) { + if (session instanceof DestroyableSession) { + ((DestroyableSession) session).destroy(); } } Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionQueue.java =================================================================== --- trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionQueue.java 2011-02-10 14:52:12 UTC (rev 21607) +++ trunk/core/impl/src/main/java/org/richfaces/application/push/impl/SessionQueue.java 2011-02-10 14:53:59 UTC (rev 21608) @@ -56,6 +56,8 @@ private final Condition available = lock.newCondition(); + private boolean active = true; + private static long getExpirationTime(Session session) { long lastAccessedTime = session.getLastAccessedTime(); if (lastAccessedTime < 0) { @@ -78,7 +80,7 @@ final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { - while (true) { + while (active) { Session first = queue.peek(); if (first == null) { available.await(); @@ -96,15 +98,19 @@ } } } + + throw new InterruptedException("Session queue is stopping"); } finally { lock.unlock(); } - } public void remove(Session session) { final ReentrantLock lock = this.lock; lock.lock(); + + checkActiveState(); + try { queue.remove(session); } finally { @@ -115,6 +121,9 @@ public void requeue(Session session, boolean addIfNotExists) { final ReentrantLock lock = this.lock; lock.lock(); + + checkActiveState(); + try { boolean exists = queue.remove(session); @@ -130,14 +139,23 @@ lock.unlock(); } } + + private void checkActiveState() { + if (!active) { + throw new IllegalStateException("Queue is not active"); + } + } - public void clear() { + public void shutdown() { final ReentrantLock lock = this.lock; lock.lock(); try { queue.clear(); + active = false; + available.signalAll(); } finally { lock.unlock(); } } + } Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/MessagingContext.java =================================================================== --- trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/MessagingContext.java 2011-02-10 14:52:12 UTC (rev 21607) +++ trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/MessagingContext.java 2011-02-10 14:53:59 UTC (rev 21608) @@ -118,12 +118,7 @@ connection.start(); } - public void stop() throws Exception { - connection.close(); - connection = null; - } - - public Connection getConnection() { + protected Connection getConnection() { if (connection == null) { throw new IllegalStateException("connection is absent"); } @@ -131,6 +126,13 @@ return connection; } + public void stop() throws Exception { + if (connection != null) { + connection.close(); + connection = null; + } + } + public Topic lookup(TopicKey topicKey) throws NamingException { Name topicName = appendToName(topicsNamespace, topicKey.getTopicName()); @@ -138,7 +140,7 @@ } public javax.jms.Session createSession() throws JMSException { - return connection.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE); + return getConnection().createSession(false, javax.jms.Session.CLIENT_ACKNOWLEDGE); } public String getSubscriptionClientId(Session session, TopicKey topicKey) { Modified: trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/PushContextImpl.java =================================================================== --- trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/PushContextImpl.java 2011-02-10 14:52:12 UTC (rev 21607) +++ trunk/core/impl/src/main/java/org/richfaces/application/push/impl/jms/PushContextImpl.java 2011-02-10 14:53:59 UTC (rev 21608) @@ -105,17 +105,17 @@ } public void destroy() { - if (messagingContext != null) { + if (pushHandlerImpl != null) { try { - messagingContext.stop(); + pushHandlerImpl.destroy(); } catch (Exception e) { LOGGER.error(e.getMessage(), e); } } - if (pushHandlerImpl != null) { + if (messagingContext != null) { try { - pushHandlerImpl.destroy(); + messagingContext.stop(); } catch (Exception e) { LOGGER.error(e.getMessage(), e); }