Author: rhauch Date: 2008-07-09 10:57:28 -0400 (Wed, 09 Jul 2008) New Revision: 346 Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/RepositoryI18n.java trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepository.java trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConfig.java trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConnection.java trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederationService.java trunk/dna-repository/src/main/resources/org/jboss/dna/repository/RepositoryI18n.properties trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederatedRepositoryConnectionTest.java trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederationServiceTest.java trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/AbstractCommandExecutor.java trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/CommandExecutor.java Log: DNA-115 - Create federation service http://jira.jboss.com/jira/browse/DNA-115 Updated the federation service, federated repository and federated repository configuration with more implementation and testing. Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/RepositoryI18n.java =================================================================== --- trunk/dna-repository/src/main/java/org/jboss/dna/repository/RepositoryI18n.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/java/org/jboss/dna/repository/RepositoryI18n.java 2008-07-09 14:57:28 UTC (rev 346) @@ -121,6 +121,7 @@ public static I18n canceledSequencingXmlDocument; public static I18n warningSequencingXmlDocument; + // Federation public static I18n interruptedWhileConnectingToFederationConfigurationRepository; public static I18n interruptedWhileUsingFederationConfigurationRepository; public static I18n interruptedWhileClosingConnectionToFederationConfigurationRepository; @@ -131,6 +132,7 @@ public static I18n unableToAuthenticateConnectionToFederatedRepository; public static I18n repositoryHasBeenShutDown; public static I18n repositoryPathInFederationBindingIsNotAbsolute; + public static I18n errorReadingMergePlan; static { try { Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepository.java =================================================================== --- trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepository.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepository.java 2008-07-09 14:57:28 UTC (rev 346) @@ -30,8 +30,10 @@ import net.jcip.annotations.ThreadSafe; import org.jboss.dna.common.util.ArgCheck; import org.jboss.dna.repository.RepositoryI18n; +import org.jboss.dna.repository.federation.impl.FederatingCommandExecutor; import org.jboss.dna.repository.services.AbstractServiceAdministrator; import org.jboss.dna.repository.services.ServiceAdministrator; +import org.jboss.dna.spi.graph.commands.executor.CommandExecutor; import org.jboss.dna.spi.graph.connection.ExecutionEnvironment; import org.jboss.dna.spi.graph.connection.RepositoryConnection; import org.jboss.dna.spi.graph.connection.RepositoryConnectionFactories; @@ -100,7 +102,7 @@ private final ExecutionEnvironment env; private final RepositoryConnectionFactories connectionFactories; private FederatedRepositoryConfig config; - private final AtomicInteger openConnections = new AtomicInteger(0); + private final AtomicInteger openExecutors = new AtomicInteger(0); private final CountDownLatch shutdownLatch = new CountDownLatch(1); private final AtomicBoolean shutdownRequested = new AtomicBoolean(false); private final CopyOnWriteArrayList<RepositorySourceListener> listeners = new CopyOnWriteArrayList<RepositorySourceListener>(); @@ -170,7 +172,7 @@ */ protected synchronized void shutdownRepository() { this.shutdownRequested.set(true); - if (this.openConnections.get() <= 0) shutdownLatch.countDown(); + if (this.openExecutors.get() <= 0) shutdownLatch.countDown(); } /** @@ -194,7 +196,7 @@ * @return true if terminated, or false otherwise */ protected boolean isTerminated() { - return this.openConnections.get() != 0; + return this.openExecutors.get() != 0; } /** @@ -273,22 +275,36 @@ } /** - * Called by {@link FederatedRepositoryConnection#FederatedRepositoryConnection(FederatedRepository, String) - * FederatedRepositoryConnection constructor}. + * Called by + * {@link FederatedRepositoryConnection#execute(ExecutionEnvironment, org.jboss.dna.spi.graph.commands.GraphCommand...)}. * - * @param federatedRepositoryConnection + * @param env the execution environment in which the executor will be run; may not be null + * @param sourceName the name of the {@link RepositorySource} that is making use of this executor; may not be null or empty + * @return the executor */ - /*package*/void register( FederatedRepositoryConnection federatedRepositoryConnection ) { - openConnections.incrementAndGet(); + protected CommandExecutor getExecutor( ExecutionEnvironment env, + String sourceName ) { + FederatedRepositoryConfig config = this.getConfiguration(); + return new FederatingCommandExecutor(env, sourceName, config.getCacheRegion(), config.getRegions(), + getConnectionFactories()); } /** + * Called by {@link FederatedRepositoryConnection#FederatedRepositoryConnection(FederatedRepository, String)}. + * + * @param connection the connection being opened + */ + /*package*/void register( FederatedRepositoryConnection connection ) { + openExecutors.incrementAndGet(); + } + + /** * Called by {@link FederatedRepositoryConnection#close()}. * - * @param federatedRepositoryConnection + * @param connection the connection being closed */ - /*package*/void unregister( FederatedRepositoryConnection federatedRepositoryConnection ) { - if (openConnections.decrementAndGet() <= 0 && shutdownRequested.get()) { + /*package*/void unregister( FederatedRepositoryConnection connection ) { + if (openExecutors.decrementAndGet() <= 0 && shutdownRequested.get()) { // Last connection, so turn out the lights ... shutdownLatch.countDown(); } Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConfig.java =================================================================== --- trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConfig.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConfig.java 2008-07-09 14:57:28 UTC (rev 346) @@ -41,6 +41,7 @@ @Immutable public class FederatedRepositoryConfig { + private final FederatedRegion cacheRegion; private final List<FederatedRegion> regions; private final Problems problems; private final String name; @@ -50,20 +51,27 @@ * Create a federated repository instance, as managed by the supplied {@link FederationService}. * * @param repositoryName the name of the repository + * @param cacheRegion the region used for the cache; may not be null * @param regions the federated regions; may not be null * @param defaultCachePolicy the default cache policy for this repository; may be null * @throws IllegalArgumentException if the name is null or is blank */ public FederatedRepositoryConfig( String repositoryName, + FederatedRegion cacheRegion, Iterable<FederatedRegion> regions, CachePolicy defaultCachePolicy ) { ArgCheck.isNotEmpty(repositoryName, "repositoryName"); + ArgCheck.isNotNull(cacheRegion, "cacheRegion"); this.name = repositoryName; this.problems = new ThreadSafeProblems(); this.defaultCachePolicy = defaultCachePolicy; + this.cacheRegion = cacheRegion; List<FederatedRegion> regionList = new ArrayList<FederatedRegion>(); for (FederatedRegion region : regions) { - if (region != null && !regionList.contains(region)) regionList.add(region); + if (region == null) continue; + if (!regionList.contains(region)) { + regionList.add(region); + } } this.regions = Collections.unmodifiableList(regionList); ArgCheck.isNotEmpty(this.regions, "regions"); @@ -89,6 +97,16 @@ } /** + * Get the region that defines the cache for this repository. This region does not exist in the {@link #getRegions() list of + * source regions}. + * + * @return the region used for caching; never null + */ + public FederatedRegion getCacheRegion() { + return cacheRegion; + } + + /** * Return the unmodifiable list of bindings. * * @return the bindings Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConnection.java =================================================================== --- trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConnection.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederatedRepositoryConnection.java 2008-07-09 14:57:28 UTC (rev 346) @@ -27,8 +27,8 @@ import net.jcip.annotations.ThreadSafe; import org.jboss.dna.repository.RepositoryI18n; import org.jboss.dna.spi.cache.CachePolicy; -import org.jboss.dna.spi.graph.commands.CompositeCommand; import org.jboss.dna.spi.graph.commands.GraphCommand; +import org.jboss.dna.spi.graph.commands.executor.CommandExecutor; import org.jboss.dna.spi.graph.connection.ExecutionEnvironment; import org.jboss.dna.spi.graph.connection.RepositoryConnection; import org.jboss.dna.spi.graph.connection.RepositorySourceException; @@ -113,31 +113,19 @@ * {@inheritDoc} */ public void execute( ExecutionEnvironment env, - GraphCommand... commands ) throws RepositorySourceException { + GraphCommand... commands ) throws RepositorySourceException, InterruptedException { if (!this.repository.getAdministrator().isStarted()) { throw new RepositorySourceException(RepositoryI18n.repositoryHasBeenShutDown.text(this.repository.getName())); } if (commands == null || commands.length == 0) return; - FederatedRepositoryConfig config = this.repository.getConfiguration(); + CommandExecutor executor = this.repository.getExecutor(env, sourceName); + assert executor != null; for (GraphCommand command : commands) { - if (command == null) continue; - executeCommand(env, command, config); + executor.execute(command); } } - protected void executeCommand( ExecutionEnvironment env, - GraphCommand command, - FederatedRepositoryConfig config ) { - if (command instanceof CompositeCommand) { - CompositeCommand composite = (CompositeCommand)command; - for (GraphCommand nestedCommand : composite) { - if (nestedCommand == null) continue; - executeCommand(env, nestedCommand, config); - } - } - } - /** * {@inheritDoc} */ Modified: trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederationService.java =================================================================== --- trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederationService.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/java/org/jboss/dna/repository/federation/FederationService.java 2008-07-09 14:57:28 UTC (rev 346) @@ -400,15 +400,25 @@ // Get the repository node ... BasicGetNodeCommand getRepository = new BasicGetNodeCommand(repositoryNode); + // Add a command to get the region defining the cache ... + Path pathToCacheRegion = pathFactory.create(repositoryNode, nameFactory.create("dna:cache")); + BasicGetNodeCommand getCacheRegion = new BasicGetNodeCommand(pathToCacheRegion); + // Get the regions for the repository ... Path regionsNode = pathFactory.create(repositoryNode, nameFactory.create("dna:regions")); BasicGetChildrenCommand getRegions = new BasicGetChildrenCommand(regionsNode); - configurationConnection.execute(env, getRepository, getRegions); + configurationConnection.execute(env, getRepository, getCacheRegion, getRegions); if (getRepository.hasError()) { throw new FederationException(RepositoryI18n.federatedRepositoryCannotBeFound.text(repositoryName)); } + if (getCacheRegion.hasError()) { + I18n msg = RepositoryI18n.requiredNodeDoesNotExistRelativeToNode; + throw new FederationException(msg.text("dna:regions", configurationRoot)); + } + FederatedRegion cacheRegion = createRegion(getCacheRegion.getPath(), getCacheRegion.getProperties(), problems); + // Build the commands to get each of the region branches ... List<FederatedRegion> regions = new LinkedList<FederatedRegion>(); if (getRegions.hasNoError() && !getRegions.getChildren().isEmpty()) { @@ -417,6 +427,7 @@ final Path pathToSource = pathFactory.create(regionsNode, child); commands.add(new BasicGetNodeCommand(pathToSource)); } + // Now execute these commands ... configurationConnection.execute(env, commands); // Iterate over each region node obtained ... @@ -442,7 +453,7 @@ cachePolicy.setTimeToExpire(longFactory.create(timeToExpireProperty.getValues().next())); } CachePolicy defaultCachePolicy = cachePolicy.isEmpty() ? null : cachePolicy.getUnmodifiable(); - return new FederatedRepositoryConfig(repositoryName, regions, defaultCachePolicy); + return new FederatedRepositoryConfig(repositoryName, cacheRegion, regions, defaultCachePolicy); } catch (InvalidPathException err) { I18n msg = RepositoryI18n.federatedRepositoryCannotBeFound; throw new FederationException(msg.text(repositoryName)); Modified: trunk/dna-repository/src/main/resources/org/jboss/dna/repository/RepositoryI18n.properties =================================================================== --- trunk/dna-repository/src/main/resources/org/jboss/dna/repository/RepositoryI18n.properties 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/main/resources/org/jboss/dna/repository/RepositoryI18n.properties 2008-07-09 14:57:28 UTC (rev 346) @@ -111,3 +111,4 @@ unableToAuthenticateConnectionToFederatedRepository = Unable to authenticate "{1}" for repository "{0}" repositoryHasBeenShutDown = The "{0}" repository has been shut down and may no longer be used. repositoryPathInFederationBindingIsNotAbsolute = The repository path in a federation binding must be absolute, but was "{0}" +errorReadingMergePlan = Error while reading merge plan for {0} \ No newline at end of file Modified: trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederatedRepositoryConnectionTest.java =================================================================== --- trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederatedRepositoryConnectionTest.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederatedRepositoryConnectionTest.java 2008-07-09 14:57:28 UTC (rev 346) @@ -31,6 +31,7 @@ import java.util.concurrent.TimeUnit; import org.jboss.dna.repository.services.ServiceAdministrator; import org.jboss.dna.spi.graph.commands.GraphCommand; +import org.jboss.dna.spi.graph.commands.executor.CommandExecutor; import org.jboss.dna.spi.graph.connection.ExecutionEnvironment; import org.jboss.dna.spi.graph.connection.RepositorySourceException; import org.jboss.dna.spi.graph.connection.RepositorySourceListener; @@ -80,7 +81,7 @@ } @Test( expected = RepositorySourceException.class ) - public void shouldFailExecutionIfRepositoryAdminsIsNotStarted() { + public void shouldFailExecutionIfRepositoryAdminsIsNotStarted() throws Exception { stub(repositoryAdmin.isStarted()).toReturn(false); ExecutionEnvironment env = mock(ExecutionEnvironment.class); GraphCommand command = mock(GraphCommand.class); @@ -88,7 +89,7 @@ } @Test - public void shouldReturnImmediatelyWhenExecutingNullOrEmptyCommandArray() { + public void shouldReturnImmediatelyWhenExecutingNullOrEmptyCommandArray() throws Exception { stub(repositoryAdmin.isStarted()).toReturn(true); ExecutionEnvironment env = mock(ExecutionEnvironment.class); connection.execute(env, (GraphCommand[])null); @@ -98,9 +99,11 @@ } @Test - public void shouldSkipNullCommandReferencesWhenExecuting() { + public void shouldSkipNullCommandReferencesWhenExecuting() throws Exception { stub(repositoryAdmin.isStarted()).toReturn(true); ExecutionEnvironment env = mock(ExecutionEnvironment.class); + CommandExecutor executor = mock(CommandExecutor.class); + stub(repository.getExecutor(env, sourceName)).toReturn(executor); connection.execute(env, new GraphCommand[] {null, null, null}); verify(repositoryAdmin, times(1)).isStarted(); } Modified: trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederationServiceTest.java =================================================================== --- trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederationServiceTest.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-repository/src/test/java/org/jboss/dna/repository/federation/FederationServiceTest.java 2008-07-09 14:57:28 UTC (rev 346) @@ -250,6 +250,9 @@ String fedReposPath = "/reposX/dna:repositories/fed repos/"; configRepository.setProperty(env, fedReposPath, TIME_TO_CACHE, "10000"); configRepository.setProperty(env, fedReposPath, TIME_TO_EXPIRE, "20000"); + configRepository.setProperty(env, fedReposPath + "dna:cache", PATH_IN_REPOSITORY, "/ca/cb/cc"); + configRepository.setProperty(env, fedReposPath + "dna:cache", PATH_IN_SOURCE, "/cx/cy"); + configRepository.setProperty(env, fedReposPath + "dna:cache", SOURCE_NAME, "cache source"); configRepository.setProperty(env, fedReposPath + "dna:regions/region1", PATH_IN_REPOSITORY, "/a/b/c"); configRepository.setProperty(env, fedReposPath + "dna:regions/region1", PATH_IN_SOURCE, "/sx/sy"); configRepository.setProperty(env, fedReposPath + "dna:regions/region1", SOURCE_NAME, "source A"); @@ -272,6 +275,10 @@ assertThat(repository.getName(), is("fed repos")); assertThat(repository.getConfiguration().getDefaultCachePolicy().getTimeToCache(), is(10000l)); assertThat(repository.getConfiguration().getDefaultCachePolicy().getTimeToExpire(), is(20000l)); + assertThat(repository.getConfiguration().getCacheRegion().getRegionName(), is("cache")); + assertThat(repository.getConfiguration().getCacheRegion().getPathInRepository(), is(pathFactory.create("/ca/cb/cc"))); + assertThat(repository.getConfiguration().getCacheRegion().getPathInSource(), is(pathFactory.create("/cx/cy"))); + assertThat(repository.getConfiguration().getCacheRegion().getSourceName(), is("cache source")); assertThat(repository.getConfiguration().getRegions().get(0).getRegionName(), is("region1")); assertThat(repository.getConfiguration().getRegions().get(0).getPathInRepository(), is(pathFactory.create("/a/b/c"))); assertThat(repository.getConfiguration().getRegions().get(0).getPathInSource(), is(pathFactory.create("/sx/sy"))); Modified: trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/AbstractCommandExecutor.java =================================================================== --- trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/AbstractCommandExecutor.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/AbstractCommandExecutor.java 2008-07-09 14:57:28 UTC (rev 346) @@ -224,4 +224,13 @@ public void execute( RecordBranchCommand command ) throws RepositorySourceException, InterruptedException { } + /** + * {@inheritDoc} + * + * @see org.jboss.dna.spi.graph.commands.executor.CommandExecutor#close() + */ + @SuppressWarnings( "unused" ) + public void close() throws InterruptedException { + } + } Modified: trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/CommandExecutor.java =================================================================== --- trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/CommandExecutor.java 2008-07-09 14:54:59 UTC (rev 345) +++ trunk/dna-spi/src/main/java/org/jboss/dna/spi/graph/commands/executor/CommandExecutor.java 2008-07-09 14:57:28 UTC (rev 346) @@ -140,4 +140,10 @@ */ void execute( MoveBranchCommand command ) throws RepositorySourceException, InterruptedException; + /** + * Close this executor, allowing it to clean up any open resources. + * + * @throws InterruptedException if the thread is interrupted while the connection is being closed + */ + void close() throws InterruptedException; }