X-Git-Url: https://git.argeo.org/?a=blobdiff_plain;f=server%2Fruntime%2Forg.argeo.server.jcr%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fjcr%2FThreadBoundJcrSessionFactory.java;h=003bcc69b552fd956e5db28b8efd27e72f81237b;hb=774e4da3fb247a95d327674045af5b5ab2a1a62a;hp=aaafdc26a704f2781c11164c07aec708a3a00769;hpb=2f1a4e0b8313d3872f7cbc73cea9f0d56cfe4fe6;p=lgpl%2Fargeo-commons.git diff --git a/server/runtime/org.argeo.server.jcr/src/main/java/org/argeo/jcr/ThreadBoundJcrSessionFactory.java b/server/runtime/org.argeo.server.jcr/src/main/java/org/argeo/jcr/ThreadBoundJcrSessionFactory.java index aaafdc26a..003bcc69b 100644 --- a/server/runtime/org.argeo.server.jcr/src/main/java/org/argeo/jcr/ThreadBoundJcrSessionFactory.java +++ b/server/runtime/org.argeo.server.jcr/src/main/java/org/argeo/jcr/ThreadBoundJcrSessionFactory.java @@ -17,11 +17,15 @@ package org.argeo.jcr; import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; import java.util.List; +import java.util.Map; import javax.jcr.LoginException; import javax.jcr.Repository; @@ -32,21 +36,15 @@ import javax.jcr.SimpleCredentials; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.argeo.ArgeoException; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.FactoryBean; /** Proxy JCR sessions and attach them to calling threads. */ -public class ThreadBoundJcrSessionFactory implements FactoryBean, - DisposableBean { +public abstract class ThreadBoundJcrSessionFactory { private final static Log log = LogFactory .getLog(ThreadBoundJcrSessionFactory.class); private Repository repository; - private final List activeSessions = Collections - .synchronizedList(new ArrayList()); private ThreadLocal session = new ThreadLocal(); - private boolean destroying = false; private final Session proxiedSession; /** If workspace is null, default will be used. */ private String workspace = null; @@ -55,6 +53,15 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, private String defaultPassword = "demo"; private Boolean forceDefaultCredentials = false; + private boolean active = true; + + // monitoring + private final List threads = Collections + .synchronizedList(new ArrayList()); + private final Map activeSessions = Collections + .synchronizedMap(new HashMap()); + private MonitoringThread monitoringThread; + public ThreadBoundJcrSessionFactory() { Class[] interfaces = { Session.class }; proxiedSession = (Session) Proxy.newProxyInstance(getClass() @@ -63,7 +70,18 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, } /** Logs in to the repository using various strategies. */ - protected Session login() { + protected synchronized Session login() { + if (!isActive()) + throw new ArgeoException("Thread bound session factory inactive"); + + // discard session previously attached to this thread + Thread thread = Thread.currentThread(); + if (activeSessions.containsKey(thread.getId())) { + Session oldSession = activeSessions.remove(thread.getId()); + oldSession.logout(); + session.remove(); + } + Session newSession = null; // first try to login without credentials, assuming the underlying login // module will have dealt with authentication (typically using Spring @@ -89,11 +107,15 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, throw new ArgeoException("Cannot log in to repository", e); } + session.set(newSession); // Log and monitor new session if (log.isTraceEnabled()) log.trace("Logged in to JCR session " + newSession + "; userId=" + newSession.getUserID()); - activeSessions.add(newSession); + + // monitoring + activeSessions.put(thread.getId(), newSession); + threads.add(thread); return newSession; } @@ -101,18 +123,68 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, return proxiedSession; } - public void destroy() throws Exception { + public void init() throws Exception { + monitoringThread = new MonitoringThread(); + monitoringThread.start(); + } + + public synchronized void dispose() throws Exception { + if (activeSessions.size() == 0) + return; + if (log.isDebugEnabled()) log.debug("Cleaning up " + activeSessions.size() + " active JCR sessions..."); - destroying = true; - for (Session sess : activeSessions) { + deactivate(); + for (Session sess : activeSessions.values()) { sess.logout(); } activeSessions.clear(); } + protected Boolean isActive() { + return active; + } + + protected synchronized void deactivate() { + active = false; + notifyAll(); + } + + protected synchronized void removeSession(Thread thread) { + if (!isActive()) + return; + activeSessions.remove(thread.getId()); + threads.remove(thread); + } + + protected synchronized void cleanDeadThreads() { + if (!isActive()) + return; + Iterator it = threads.iterator(); + while (it.hasNext()) { + Thread thread = it.next(); + if (!thread.isAlive() && isActive()) { + if (activeSessions.containsKey(thread.getId())) { + Session session = activeSessions.get(thread.getId()); + activeSessions.remove(thread.getId()); + session.logout(); + if (log.isDebugEnabled()) + log.debug("Cleaned up JCR session (userID=" + + session.getUserID() + ") from dead thread " + + thread.getId()); + } + it.remove(); + } + } + try { + wait(1000); + } catch (InterruptedException e) { + // silent + } + } + public Class getObjectType() { return Session.class; } @@ -121,6 +193,15 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, return true; } + /** + * Called before a method is actually called, allowing to check the session + * or re-login it (e.g. if authentication has changed). The default + * implementation returns the session. + */ + protected Session preCall(Session session) { + return session; + } + public void setRepository(Repository repository) { this.repository = repository; } @@ -144,7 +225,7 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, protected class JcrSessionInvocationHandler implements InvocationHandler { public Object invoke(Object proxy, Method method, Object[] args) - throws Throwable { + throws Throwable, RepositoryException { Session threadSession = session.get(); if (threadSession == null) { if ("logout".equals(method.getName()))// no need to login @@ -152,19 +233,45 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, else if ("toString".equals(method.getName()))// maybe logging return "Uninitialized Argeo thread bound JCR session"; threadSession = login(); - session.set(threadSession); } - Object ret = method.invoke(threadSession, args); + preCall(threadSession); + Object ret; + try { + ret = method.invoke(threadSession, args); + } catch (InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof RepositoryException) + throw (RepositoryException) cause; + else + throw cause; + } if ("logout".equals(method.getName())) { session.remove(); - if (!destroying) - activeSessions.remove(threadSession); + Thread thread = Thread.currentThread(); + removeSession(thread); if (log.isTraceEnabled()) - log.trace("Logged out from JCR session " + threadSession - + "; userId=" + threadSession.getUserID()); + log.trace("Logged out JCR session (userId=" + + threadSession.getUserID() + ") on thread " + + thread.getId()); } return ret; } } + + /** Monitors registered thread in order to clean up dead ones. */ + private class MonitoringThread extends Thread { + + public MonitoringThread() { + super("ThreadBound JCR Session Monitor"); + } + + @Override + public void run() { + while (isActive()) { + cleanDeadThreads(); + } + } + + } }