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=9dee3a078e23f1dc4600f99895af1c32e88e5b0b;hb=1d5afdce3e91054f07ddd3c98309c363b4cf1d46;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..9dee3a078 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 @@ -1,5 +1,5 @@ /* - * Copyright (C) 2010 Mathieu Baudier + * Copyright (C) 2007-2012 Mathieu Baudier * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,15 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - 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 +35,17 @@ 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()); + /** can be injected as list, only used if repository is null */ + private List repositories; 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 +54,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,14 +71,25 @@ 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 // Security) if (!forceDefaultCredentials) try { - newSession = repository.login(workspace); + newSession = repository().login(workspace); } catch (LoginException e1) { log.warn("Cannot login without credentials: " + e1.getMessage()); // invalid credentials, go to the next step @@ -84,16 +103,20 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, try { SimpleCredentials sc = new SimpleCredentials(defaultUsername, defaultPassword.toCharArray()); - newSession = repository.login(sc, workspace); + newSession = repository().login(sc, workspace); } catch (RepositoryException e) { 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 +124,68 @@ public class ThreadBoundJcrSessionFactory implements FactoryBean, return proxiedSession; } - public void destroy() throws Exception { - if (log.isDebugEnabled()) - log.debug("Cleaning up " + activeSessions.size() + public void init() throws Exception { + monitoringThread = new MonitoringThread(); + monitoringThread.start(); + } + + public synchronized void dispose() throws Exception { + if (activeSessions.size() == 0) + return; + + if (log.isTraceEnabled()) + log.trace("Cleaning up " + activeSessions.size() + " active JCR sessions..."); - destroying = true; - for (Session sess : activeSessions) { - sess.logout(); + deactivate(); + for (Session sess : activeSessions.values()) { + JcrUtils.logoutQuietly(sess); } 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.isTraceEnabled()) + log.trace("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,10 +194,45 @@ 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; + } + + protected Repository repository() { + if (repository != null) + return repository; + if (repositories != null) { + // hardened for OSGi dynamic services + Iterator it = repositories.iterator(); + if (it.hasNext()) + return it.next(); + } + throw new ArgeoException("No repository injected"); + } + + // /** Useful for declarative registration of OSGi services (blueprint) */ + // public void register(Repository repository, Map params) { + // this.repository = repository; + // } + // + // /** Useful for declarative registration of OSGi services (blueprint) */ + // public void unregister(Repository repository, Map params) { + // this.repository = null; + // } + public void setRepository(Repository repository) { this.repository = repository; } + public void setRepositories(List repositories) { + this.repositories = repositories; + } + public void setDefaultUsername(String defaultUsername) { this.defaultUsername = defaultUsername; } @@ -144,7 +252,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 +260,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(); + } + } + + } }