[Mulgara-svn] r1344 - trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed

[ronald at mulgara.org]

Author: ronald
Date: 2008-10-23 06:11:16 -0700 (Thu, 23 Oct 2008)
New Revision: 1344

Modified:
   trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/Delegator.java
   trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolver.java
   trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolverFactory.java
   trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/NetworkDelegator.java
Log:
Fix closing of sessions and plug memory leak. The remote sessions and their
factories are now closed at the end of every transaction, instead of at the
time the resolver-factory is closed (which usually isn't until mulgara is
shut down).

Modified: trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/Delegator.java
===================================================================
--- trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/Delegator.java	2008-10-23 13:11:09 UTC (rev 1343)
+++ trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/Delegator.java	2008-10-23 13:11:16 UTC (rev 1344)
@@ -59,4 +59,9 @@
    * @throws QueryException There was an error removing data at the remote end.
    */
   public void remove(long model, Statements statements) throws ResolverException, QueryException;
+
+  /**
+   * Close the delegator.
+   */
+  public void close();
 }

Modified: trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolver.java
===================================================================
--- trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolver.java	2008-10-23 13:11:09 UTC (rev 1343)
+++ trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolver.java	2008-10-23 13:11:16 UTC (rev 1344)
@@ -15,7 +15,9 @@
 // Java 2 standard packages
 import java.net.*;
 import java.util.*;
+import javax.transaction.xa.XAException;
 import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
 
 // Third party packages
 import org.apache.log4j.Logger;
@@ -51,14 +53,10 @@
   /** The delegator that resolves the constraint on another server.  */
   private final Delegator delegator;
 
-  /** A map of servers to sessions.  This acts as a cache, and also so we may close the sessions.  */
-  private Map<URI,Session> serverToSession = new HashMap<URI,Session>();
+  /** our xa-resource */
+  private final XAResource xares;
 
-  /** A collections of session factories.  This is so we may close these factories.  */
-  private Collection<SessionFactory> sessionFactories = new HashSet<SessionFactory>();
 
-
-
   /**
    * Construct a Distributed Resolver.
    * @param resolverSession the session this resolver is associated with.
@@ -73,6 +71,16 @@
     if (resolverSession == null) throw new IllegalArgumentException( "Null \"resolverSession\" parameter");
 
     delegator = new NetworkDelegator(resolverSession);
+
+    xares = new DummyXAResource() {
+      public void commit(Xid xid, boolean onePhase) throws XAException {
+        delegator.close();
+      }
+
+      public void rollback(Xid xid) throws XAException {
+        delegator.close();
+      }
+    };
   }
 
 
@@ -87,18 +95,13 @@
 
   /**
    * Expose a callback object for enlistment by a transaction manager.
-   * Uses a dumy xa resource for the moment, but may need to create a fully
-   * functional xa resource which is mapped to a session.
    *
    * @return an {@link XAResource} that can be used by a transaction manager to
    *   coordinate this resolver's participation in a distributed transaction.
-   *   For now this is a {@link DummyXAResource} with a 10 second transaction timeout
    * @see javax.resource.spi.ManagedConnection#getXAResource
    */
   public XAResource getXAResource() {
-    return new DummyXAResource(
-      10  // seconds before transaction timeout
-    );
+    return xares;
   }
 
 
@@ -147,7 +150,7 @@
     if (constraint == null) throw new IllegalArgumentException();
     ConstraintElement modelElement = constraint.getElement(3);
     if (!(modelElement instanceof LocalNode)) throw new QueryException("Constraint not set to a distributed model.");
-    
+
     try {
       return delegator.resolve(constraint, (LocalNode)modelElement);
     } catch (ResolverException re) {
@@ -156,28 +159,7 @@
   }
 
 
-  /**
-   * Close all sessions and factories used by this resolver.
-   */
-  public void close() {
-    for (Session s: serverToSession.values()) {
-      try {
-        s.close();
-      } catch (QueryException qe) {
-        logger.error("Exception while closing session", qe);
-      }
-    }
-    for (SessionFactory sf: sessionFactories) {
-      try {
-        sf.close();
-      } catch (QueryException qe) {
-        logger.error("Exception while closing session", qe);
-      }
-    }
-  }
-
-
   public void abort() {
-    // no-op
+    delegator.close();
   }
 }

Modified: trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolverFactory.java
===================================================================
--- trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolverFactory.java	2008-10-23 13:11:09 UTC (rev 1343)
+++ trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/DistributedResolverFactory.java	2008-10-23 13:11:16 UTC (rev 1344)
@@ -44,9 +44,6 @@
   /** Logger. */
   private static Logger logger = Logger.getLogger(DistributedResolverFactory.class);
 
-  /** Collection of constructed resolvers. */
-  private Collection<DistributedResolver> openResolvers = new HashSet<DistributedResolver>();
-
   /** Protocols which are handled by the served resolver. */
   private static final Set<String> protocols = new HashSet<String>(Arrays.asList("rmi"));
 
@@ -71,8 +68,6 @@
    * {@inheritDoc ResolverFactory}
    */
   public void close() {
-    logger.debug("Closing distributed resolvers");
-    for (DistributedResolver r: openResolvers) r.close();
   }
 
   /**
@@ -125,11 +120,11 @@
   ) throws ResolverFactoryException {
 
     if (resolverSession == null) throw new IllegalArgumentException("No session provided for the resolver!");
-    logger.debug("Creating new distributed resolver");
-    if (canWrite) logger.debug("Expecting to write to distributed resolver.");
-    DistributedResolver r = new DistributedResolver(resolverSession);
-    openResolvers.add(r);
-    return r;
+    if (logger.isDebugEnabled()) {
+      logger.debug("Creating new distributed resolver");
+      if (canWrite) logger.debug("Expecting to write to distributed resolver.");
+    }
+    return new DistributedResolver(resolverSession);
   }
 
 

Modified: trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/NetworkDelegator.java
===================================================================
--- trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/NetworkDelegator.java	2008-10-23 13:11:09 UTC (rev 1343)
+++ trunk/src/jar/resolver-distributed/java/org/mulgara/resolver/distributed/NetworkDelegator.java	2008-10-23 13:11:16 UTC (rev 1344)
@@ -327,4 +327,23 @@
     }
   }
 
+  /**
+   * Close all sessions and factories used by this delegator.
+   */
+  public void close() {
+    for (Session s: sessionCache.values()) {
+      try {
+        s.close();
+      } catch (QueryException qe) {
+        logger.error("Exception while closing session", qe);
+      }
+    }
+    for (SessionFactory sf: factoryCache) {
+      try {
+        sf.close();
+      } catch (QueryException qe) {
+        logger.error("Exception while closing session", qe);
+      }
+    }
+  }
 }