Implemented ConnectionFactory for RepositoryManagers

README.markdown

@@ -27,7 +27,7 @@ Wiring up a simple in-memory repository to the transaction manager:
 </bean>
 
 <!-- Transaction manager configuration -->
-<bean id="sesameConnectionFactory" class="org.openrdf.spring.SesameConnectionFactory">
+<bean id="sesameConnectionFactory" class="org.openrdf.spring.RepositoryConnectionFactory">
     <constructor-arg ref="sesameRepository"/>
 </bean>
 

pom.xml

@@ -75,6 +75,11 @@
             <artifactId>sesame-query</artifactId>
             <version>${sesame.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.openrdf.sesame</groupId>
+            <artifactId>sesame-repository-manager</artifactId>
+            <version>${sesame.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.openrdf.sesame</groupId>
             <artifactId>sesame-queryparser-sparql</artifactId>

src/main/java/org/openrdf/spring/RepositoryConnectionFactory.java

@@ -0,0 +1,202 @@
+package org.openrdf.spring;
+
+import org.openrdf.repository.Repository;
+import org.openrdf.repository.RepositoryConnection;
+import org.openrdf.repository.RepositoryException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.transaction.TransactionSystemException;
+
+/**
+ * <p>{@link RepositoryConnectionFactory} handles connections to the corresponding {@link Repository} and manages
+ * the transaction state (represented by {@link SesameTransactionObject}).</p>
+ * <p/>
+ * <p>This class provides methods to access <i>transactional</i> connections from the outside and is typically
+ * the only class that library users interact with.</p>
+ *
+ * @author ameingast@gmail.com
+ */
+public class RepositoryConnectionFactory implements DisposableBean, SesameConnectionFactory {
+    private final ThreadLocal<SesameTransactionObject> localTransactionObject;
+
+    private Repository repository;
+
+    public RepositoryConnectionFactory(Repository repository) {
+        this.repository = repository;
+        localTransactionObject = new ThreadLocal<SesameTransactionObject>();
+    }
+
+    /**
+     * <p>Retrieves the connection for the current transaction. This method may be called at any time as long as a
+     * transaction is active and will return the same connection-handle to the repository in the same
+     * transaction context (which is thread-local).</p>
+     *
+     * @return the {@link RepositoryConnection}
+     * @throws SesameTransactionException if
+     *                                    <ul>
+     *                                    <li>No transaction is active</li>
+     *                                    </ul>
+     */
+    @Override
+    public RepositoryConnection getConnection() {
+        SesameTransactionObject sesameTransactionObject = localTransactionObject.get();
+
+        if (sesameTransactionObject == null) {
+            throw new SesameTransactionException("No transaction active");
+        }
+
+        RepositoryConnection repositoryConnection = sesameTransactionObject.getRepositoryConnection();
+
+        try {
+            if (!repositoryConnection.isOpen()) {
+                throw new SesameTransactionException("Connection closed during transaction");
+            }
+        } catch (RepositoryException e) {
+            throw new SesameTransactionException(e);
+        }
+
+        return repositoryConnection;
+    }
+
+    /**
+     * <p>Closes the connection and cleans up the (thread-local) state for the current transaction.</p>
+     * <p/>
+     * <p>This method should not be called manually, since the connection is managed by the
+     * {@link SesameTransactionManager}.</p>
+     * <p></p>
+     *
+     * @throws SesameTransactionException if
+     *                                    <ul>
+     *                                    <li>No transaction is active</li>
+     *                                    <li>The connection could not be closed</li>
+     *                                    </ul>
+     */
+    @Override
+    public void closeConnection() {
+        SesameTransactionObject sesameTransactionObject = null;
+        RepositoryConnection repositoryConnection = null;
+
+        try {
+            sesameTransactionObject = localTransactionObject.get();
+
+            if (sesameTransactionObject == null) {
+                throw new SesameTransactionException("No transaction active");
+            }
+
+            repositoryConnection = sesameTransactionObject.getRepositoryConnection();
+
+            try {
+                if (!repositoryConnection.isOpen()) {
+                    throw new SesameTransactionException("Connection closed during transaction");
+                }
+            } catch (RepositoryException e) {
+                throw new SesameTransactionException(e);
+            }
+        } finally {
+            if (sesameTransactionObject != null && repositoryConnection != null) {
+                try {
+                    repositoryConnection.close();
+                } catch (RepositoryException e) {
+                    throw new SesameTransactionException(e);
+                }
+
+                localTransactionObject.remove();
+            }
+        }
+    }
+
+    /**
+     * <p>Shuts down the {@link Repository} if it was initialized before.</p>
+     *
+     * @throws Exception {@see Repository#shutDown}
+     */
+    @Override
+    public void destroy() throws Exception {
+        if (repository != null && repository.isInitialized()) {
+            try {
+                repository.shutDown();
+            } finally {
+                repository = null;
+            }
+        }
+    }
+
+    /**
+     * <p>Creates a new {@link SesameTransactionObject}, connects the created object
+     * to the corresponding {@link Repository} and disables auto-commit on the connection.</p>
+     * <p/>
+     * <p>This method should only be called by {@link SesameTransactionManager}.</p>
+     *
+     * @return the created transaction object representing the transaction state.
+     * @throws RepositoryException {@see Repository#getConnection}
+     */
+    @Override
+    public SesameTransactionObject createTransaction() throws RepositoryException {
+        RepositoryConnection connection = repository.getConnection();
+        connection.setAutoCommit(false);
+
+        SesameTransactionObject sesameTransactionObject = new SesameTransactionObject(connection);
+        localTransactionObject.set(sesameTransactionObject);
+
+        return sesameTransactionObject;
+    }
+
+    /**
+     * <p>Ends the active transaction by either rolling-back or committing the changes to the {@link Repository}
+     * depending on the rollback-flag.</p>
+     * <p/>
+     * <p>This method should only be called by {@link SesameTransactionManager}.</p>
+     *
+     * @param rollback if <code>true</code> the current transaction is rolled back, if <code>false</code> the pending
+     *                 changes on the connection are committed to the {@link Repository}.
+     * @throws RepositoryException        if
+     *                                    <ul>
+     *                                    <li>The changes could not be rolled back</li>
+     *                                    <li>The changes could not be committed</li>
+     *                                    </ul>
+     * @throws SesameTransactionException if
+     *                                    <ul>
+     *                                    <li>No transaction is active</li>
+     *                                    <li>The connection could not be closed</li>
+     *                                    </ul>
+     */
+    @Override
+    public void endTransaction(boolean rollback) throws RepositoryException {
+        SesameTransactionObject sesameTransactionObject = localTransactionObject.get();
+
+        if (sesameTransactionObject == null) {
+            throw new TransactionSystemException("No transaction active");
+        }
+
+        RepositoryConnection repositoryConnection = sesameTransactionObject.getRepositoryConnection();
+
+        if (!repositoryConnection.isOpen()) {
+            throw new SesameTransactionException("Connection closed during transaction");
+        }
+
+        if (rollback) {
+            repositoryConnection.rollback();
+        } else {
+            repositoryConnection.commit();
+        }
+    }
+
+    /**
+     * <p>Retrieves the current transaction state.</p>
+     * <p/>
+     * <p>This method should only be called by {@link SesameTransactionManager}.</p>
+     *
+     * @return The current transaction state in form of a {@link SesameTransactionObject}.
+     */
+    @Override
+    public SesameTransactionObject getLocalTransactionObject() {
+        return localTransactionObject.get();
+    }
+
+    @Override
+    public String toString() {
+        return "RepositoryConnectionFactory{" +
+                "repository=" + repository +
+                ", localTransactionObject=" + localTransactionObject +
+                '}';
+    }
+}

src/main/java/org/openrdf/spring/RepositoryManagerConnectionFactory.java

@@ -0,0 +1,112 @@
+package org.openrdf.spring;
+
+import org.openrdf.repository.Repository;
+import org.openrdf.repository.RepositoryConnection;
+import org.openrdf.repository.RepositoryException;
+import org.openrdf.repository.config.RepositoryConfigException;
+import org.openrdf.repository.manager.RepositoryManager;
+
+/**
+ * <p>{@link RepositoryManagerConnectionFactory} handles connections to a {@link RepositoryManager} and manages
+ * the transaction state (represented by {@link SesameTransactionObject}).</p>
+ * <p/>
+ * <p>The {@link RepositoryManager}'s repository-id is set via the property
+ * {@link RepositoryManagerConnectionFactory#setLocalRepositoryId(String)}/
+ * {@link org.openrdf.spring.RepositoryManagerConnectionFactory#getLocalRepositoryId()}.
+ * <p/>
+ * </p>It supports a single repository-id per thread; i.e. you can use the {@link RepositoryManagerConnectionFactory}
+ * with different @{link Repository}s in different threads.
+ * <p/>
+ * <p>This class provides methods to access <i>transactional</i> connections from the outside and is typically
+ * the only class that library users interact with.</p>
+ *
+ * @author ameingast@gmail.com
+ */
+public class RepositoryManagerConnectionFactory implements SesameConnectionFactory {
+    private final RepositoryManager repositoryManager;
+
+    private ThreadLocal<String> localRepositoryId;
+
+    private ThreadLocal<RepositoryConnectionFactory> localConnectionFactory;
+
+    public RepositoryManagerConnectionFactory(RepositoryManager repositoryManager) {
+        this.repositoryManager = repositoryManager;
+        localRepositoryId = new ThreadLocal<String>();
+        localConnectionFactory = new ThreadLocal<RepositoryConnectionFactory>();
+    }
+
+    @Override
+    public RepositoryConnection getConnection() {
+        initializeLocalConnectionFactory();
+
+        return localConnectionFactory.get().getConnection();
+    }
+
+    private void initializeLocalConnectionFactory() {
+        if (localConnectionFactory.get() == null) {
+            if (localRepositoryId.get() == null || localRepositoryId.get().isEmpty()) {
+                throw new RuntimeException("Local repository-id has not been initialized");
+            }
+
+            try {
+                Repository repository = repositoryManager.getRepository(localRepositoryId.get());
+                RepositoryConnectionFactory repositoryConnectionFactory = new RepositoryConnectionFactory(repository);
+
+                localConnectionFactory.set(repositoryConnectionFactory);
+            } catch (RepositoryException e) {
+                throw new RuntimeException(e);
+            } catch (RepositoryConfigException e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    @Override
+    public void closeConnection() {
+        try {
+            initializeLocalConnectionFactory();
+
+            localConnectionFactory.get().closeConnection();
+        } finally {
+            localConnectionFactory.remove();
+        }
+    }
+
+    @Override
+    public SesameTransactionObject createTransaction() throws RepositoryException {
+        initializeLocalConnectionFactory();
+
+        return localConnectionFactory.get().createTransaction();
+    }
+
+    @Override
+    public void endTransaction(boolean rollback) throws RepositoryException {
+        initializeLocalConnectionFactory();
+
+        localConnectionFactory.get().endTransaction(rollback);
+    }
+
+    @Override
+    public SesameTransactionObject getLocalTransactionObject() {
+        initializeLocalConnectionFactory();
+
+        return localConnectionFactory.get().getLocalTransactionObject();
+    }
+
+    public String getLocalRepositoryId() {
+        return localRepositoryId.get();
+    }
+
+    public void setLocalRepositoryId(String repositoryId) {
+        this.localRepositoryId.set(repositoryId);
+    }
+
+    @Override
+    public String toString() {
+        return "RepositoryManagerConnectionFactory{" +
+                "repositoryManager=" + repositoryManager +
+                ", repositoryId='" + localRepositoryId + '\'' +
+                ", connectionFactory=" + localConnectionFactory +
+                '}';
+    }
+}