Author: rhauch Date: 2009-07-20 23:14:59 -0400 (Mon, 20 Jul 2009) New Revision: 1118 Added: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/RequestException.java Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/Graph.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/BranchedMirrorProjector.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnection.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySource.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedWorkspace.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessor.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/JoinRequestProcessor.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/CompositeRequest.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidRequestException.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidWorkspaceException.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/MoveBranchRequest.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/UnsupportedRequestException.java trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/processor/RequestProcessor.java trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/AbstractFederatedRepositorySourceIntegrationTest.java trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnectionTest.java trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySourceUsingMirrorAndBranchProjectionsTest.java trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessorChannelTest.java Log: DNA-392 Fixes and improvements to the federated connector. Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/Graph.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/Graph.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/Graph.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -5451,14 +5451,23 @@ if (request instanceof ReadAllPropertiesRequest) { ReadAllPropertiesRequest read = (ReadAllPropertiesRequest)request; DateTime expires = computeExpirationTime(read); + if (read.getActualLocationOfNode() == null) { + int x = 0; + } getOrCreateNode(read.getActualLocationOfNode(), expires).setProperties(read.getPropertiesByName()); } else if (request instanceof ReadPropertyRequest) { ReadPropertyRequest read = (ReadPropertyRequest)request; DateTime expires = computeExpirationTime(read); + if (read.getActualLocationOfNode() == null) { + int x = 0; + } getOrCreateNode(read.getActualLocationOfNode(), expires).addProperty(read.getProperty()); } else if (request instanceof ReadNodeRequest) { ReadNodeRequest read = (ReadNodeRequest)request; DateTime expires = computeExpirationTime(read); + if (read.getActualLocationOfNode() == null) { + int x = 0; + } BatchResultsNode node = getOrCreateNode(read.getActualLocationOfNode(), expires); node.setProperties(read.getPropertiesByName()); node.setChildren(read.getChildren()); @@ -5467,11 +5476,17 @@ } else if (request instanceof ReadAllChildrenRequest) { ReadAllChildrenRequest read = (ReadAllChildrenRequest)request; DateTime expires = computeExpirationTime(read); + if (read.getActualLocationOfNode() == null) { + int x = 0; + } getOrCreateNode(read.getActualLocationOfNode(), expires).setChildren(read.getChildren()); } else if (request instanceof ReadBranchRequest) { ReadBranchRequest read = (ReadBranchRequest)request; DateTime expires = computeExpirationTime(read); for (Location location : read) { + if (location == null) { + int x = 0; + } BatchResultsNode node = getOrCreateNode(location, expires); node.setProperties(read.getPropertiesFor(location)); node.setChildren(read.getChildren(location)); @@ -5497,6 +5512,9 @@ } else if (request instanceof ReadNodeRequest) { ReadNodeRequest read = (ReadNodeRequest)request; DateTime expires = computeExpirationTime(read); + if (read.getActualLocationOfNode() == null) { + int x = 0; + } BatchResultsNode node = getOrCreateNode(read.getActualLocationOfNode(), expires); node.setProperties(read.getPropertiesByName()); node.setChildren(read.getChildren()); @@ -5537,7 +5555,11 @@ DateTime expirationTime ) { BatchResultsNode node = nodes.get(location); if (node == null) { + if (location == null) { + int x = 0; + } node = new BatchResultsNode(location, expirationTime); + assert location != null; assert location.getPath() != null; nodes.put(location.getPath(), node); } Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/BranchedMirrorProjector.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/BranchedMirrorProjector.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/BranchedMirrorProjector.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -120,6 +120,7 @@ public ProjectedNode project( ExecutionContext context, Location location, boolean requiresUpdate ) { + assert location != null; if (location.hasPath()) { Path path = location.getPath(); if (path.isRoot()) { Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnection.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnection.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnection.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -134,13 +134,14 @@ } protected boolean shouldProcessSynchronously( Request request ) { - if (request instanceof CompositeRequest) { - CompositeRequest composite = (CompositeRequest)request; - if (composite.size() == 1) return true; - return false; - } - // Otherwise, its just a single request ... return true; + // if (request instanceof CompositeRequest) { + // CompositeRequest composite = (CompositeRequest)request; + // if (composite.size() == 1) return true; + // return false; + // } + // // Otherwise, its just a single request ... + // return true; } /** @@ -180,7 +181,7 @@ } finally { fork.close(); } - if (!awaitAllSubtasks) requests.add(new NoMoreFederatedRequests()); + requests.add(new NoMoreFederatedRequests()); // At this point, all submitted requests have been processed/forked, so we can continue with // the join process, starting with the first submitted request. Note that the subtasks may // still be executing, but as the join process operates on a forked request, it will wait @@ -195,7 +196,7 @@ } finally { fork.close(); } - if (!awaitAllSubtasks) requests.add(new NoMoreFederatedRequests()); + requests.add(new NoMoreFederatedRequests()); } }); Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySource.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySource.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySource.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -40,6 +40,7 @@ import javax.naming.spi.ObjectFactory; import net.jcip.annotations.GuardedBy; import org.jboss.dna.common.i18n.I18n; +import org.jboss.dna.common.util.CheckArg; import org.jboss.dna.common.util.HashCode; import org.jboss.dna.graph.DnaLexicon; import org.jboss.dna.graph.ExecutionContext; @@ -387,6 +388,93 @@ } /** + * Add a federated workspace to this source. If a workspace with the supplied name already exists, it will be replaced with + * the new one. + * + * @param workspaceName the name of the new federated workspace + * @param projections the projections that should be used in the workspace + * @param isDefault true if this workspace should be used as the default workspace, or false otherwise + * @return the federated workspace + * @throws IllegalArgumentException if the workspace name or the projections reference are null + */ + public synchronized FederatedWorkspace addWorkspace( String workspaceName, + Iterable<Projection> projections, + boolean isDefault ) { + CheckArg.isNotNull(workspaceName, "workspaceName"); + CheckArg.isNotNull(projections, "projections"); + + // Check all the properties of this source ... + String name = getName(); + if (name == null) { + I18n msg = GraphI18n.namePropertyIsRequiredForFederatedRepositorySource; + throw new RepositorySourceException(getName(), msg.text("name")); + } + RepositoryContext context = getRepositoryContext(); + if (context == null) { + I18n msg = GraphI18n.federatedRepositorySourceMustBeInitialized; + throw new RepositorySourceException(getName(), msg.text("name", name)); + } + + // Now set up or get the existing components needed by the workspace ... + RepositoryConnectionFactory connectionFactory = null; + ExecutorService executor = null; + LinkedList<FederatedWorkspace> workspaces = new LinkedList<FederatedWorkspace>(); + CachePolicy defaultCachePolicy = null; + if (this.configuration != null) { + connectionFactory = this.configuration.getConnectionFactory(); + executor = this.configuration.getExecutor(); + defaultCachePolicy = this.configuration.getDefaultCachePolicy(); + for (String existingWorkspaceName : this.configuration.getWorkspaceNames()) { + if (existingWorkspaceName.equals(workspaceName)) continue; + workspaces.add(this.configuration.getWorkspace(existingWorkspaceName)); + } + } else { + connectionFactory = context.getRepositoryConnectionFactory(); + executor = Executors.newCachedThreadPool(); + } + + // Add the new workspace ... + FederatedWorkspace newWorkspace = new FederatedWorkspace(context, name, workspaceName, projections, defaultCachePolicy); + if (isDefault) { + workspaces.addFirst(newWorkspace); + } else { + workspaces.add(newWorkspace); + } + // Update the configuration ... + this.configuration = new FederatedRepository(name, connectionFactory, workspaces, defaultCachePolicy, executor); + return newWorkspace; + } + + /** + * Remove the named workspace from the repository source. + * + * @param workspaceName the name of the workspace to remove + * @return true if the workspace was removed, or false otherwise + * @throws IllegalArgumentException if the workspace name is null + */ + public synchronized boolean removeWorkspace( String workspaceName ) { + CheckArg.isNotNull(workspaceName, "workspaceName"); + if (this.configuration == null) return false; + FederatedWorkspace workspace = this.configuration.getWorkspace(workspaceName); + if (workspace == null) return false; + List<FederatedWorkspace> workspaces = new LinkedList<FederatedWorkspace>(); + for (String existingWorkspaceName : this.configuration.getWorkspaceNames()) { + if (existingWorkspaceName.equals(workspaceName)) continue; + workspaces.add(this.configuration.getWorkspace(existingWorkspaceName)); + } + RepositoryConnectionFactory connectionFactory = this.configuration.getConnectionFactory(); + ExecutorService executor = this.configuration.getExecutor(); + CachePolicy defaultCachePolicy = this.configuration.getDefaultCachePolicy(); + this.configuration = new FederatedRepository(name, connectionFactory, workspaces, defaultCachePolicy, executor); + return true; + } + + public synchronized boolean hasWorkspace( String workspaceName ) { + CheckArg.isNotNull(workspaceName, "workspaceName"); + return this.configuration != null && this.configuration.getWorkspaceNames().contains(workspaceName); + } + + /** * Instantiate the {@link Projection} described by the supplied properties. * * @param context the execution context that should be used to read the configuration; may not be null Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedWorkspace.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedWorkspace.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/FederatedWorkspace.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -71,6 +71,7 @@ CheckArg.isNotNull(repositoryContext, "repositoryContext"); CheckArg.isNotNull(sourceName, "sourceName"); CheckArg.isNotNull(workspaceName, "workspaceName"); + CheckArg.isNotNull(projections, "projections"); this.repositoryContext = repositoryContext; this.workspaceName = workspaceName; this.sourceName = sourceName; Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessor.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessor.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessor.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -72,6 +72,8 @@ import org.jboss.dna.graph.request.ReadBranchRequest; import org.jboss.dna.graph.request.ReadNodeRequest; import org.jboss.dna.graph.request.ReadPropertyRequest; +import org.jboss.dna.graph.request.RemovePropertyRequest; +import org.jboss.dna.graph.request.RenameNodeRequest; import org.jboss.dna.graph.request.Request; import org.jboss.dna.graph.request.SetPropertyRequest; import org.jboss.dna.graph.request.UnsupportedRequestException; @@ -118,7 +120,7 @@ * A pseudo Request that allows addition of a request that, when processed, will decrement a latch. Since the latch is * supplied by the submitter, this is useful when the submitter wishes to block until a particular request is processed. The * submitter merely creates a {@link CountDownLatch}, submits their real request wrapped by a BlockedRequest, and calls - * {@link CountDownLatch#await()} on the latch. When <code>await()</code> returns, the first request has been completed. + * {@link CountDownLatch#await()} on the latch. When <code>await()</code> returns, the request has been completed. * * @see ForkRequestProcessor#submit(Request, String, CountDownLatch) * @see ForkRequestProcessor#submitAndAwait(Request, String) @@ -143,6 +145,27 @@ public boolean isCancelled() { return original.isCancelled(); } + + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.Request#setError(java.lang.Throwable) + */ + @Override + public void setError( Throwable error ) { + original.setError(error); + if (this.latch != null) this.latch.countDown(); + } + + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.Request#hasError() + */ + @Override + public boolean hasError() { + return original.hasError(); + } } /** @@ -172,6 +195,7 @@ protected Future<String> future; /** Flag that defines whether the channel has processed all requests */ protected final AtomicBoolean done = new AtomicBoolean(false); + protected Throwable compositeError = null; /** * Create a new channel that operates against the supplied source. @@ -225,6 +249,27 @@ public void cancel() { done.set(true); } + + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.Request#setError(java.lang.Throwable) + */ + @Override + public void setError( Throwable error ) { + compositeError = error; + super.setError(error); + } + + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.Request#hasError() + */ + @Override + public boolean hasError() { + return compositeError != null || super.hasError(); + } }; } @@ -1077,6 +1122,41 @@ /** * {@inheritDoc} * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.RemovePropertyRequest) + */ + @Override + public void process( RemovePropertyRequest request ) { + // Figure out where this request is projected ... + ProjectedNode projectedNode = project(request.from(), request.inWorkspace(), request, true); + if (projectedNode == null) return; + + // Create the federated request ... + FederatedRequest federatedRequest = new FederatedRequest(request); + + // Any non-read request should be submitted to the first ProxyNode ... + while (projectedNode != null) { + if (projectedNode.isProxy()) { + ProxyNode proxy = projectedNode.asProxy(); + // Create and submit a request for the projection ... + RemovePropertyRequest pushDownRequest = new RemovePropertyRequest(proxy.location(), proxy.workspaceName(), + request.propertyName()); + federatedRequest.add(pushDownRequest, proxy.isSameLocationAsOriginal(), false, proxy.projection()); + + // Submit the requests for processing and then STOP ... + submit(federatedRequest); + return; + } + assert projectedNode.isPlaceholder(); + projectedNode = projectedNode.next(); + } + // Unable to perform this update ... + String msg = GraphI18n.unableToUpdatePlaceholder.text(readable(request.from()), request.inWorkspace(), getSourceName()); + request.setError(new UnsupportedRequestException(msg)); + } + + /** + * {@inheritDoc} + * * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.UpdatePropertiesRequest) */ @Override @@ -1147,6 +1227,33 @@ /** * {@inheritDoc} * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.DeleteChildrenRequest) + */ + @Override + public void process( DeleteChildrenRequest request ) { + // Figure out where this request is projected ... + ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, true); + if (projectedNode == null) return; + + // Create the federated request ... + FederatedRequest federatedRequest = new FederatedRequest(request); + + // A delete should be executed against any ProxyNode that applies ... + FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); + boolean submit = deleteBranch(federatedRequest, projectedNode, workspace, getExecutionContext(), false); + if (submit) { + // Submit the requests for processing and then STOP ... + submit(federatedRequest); + } else { + // Unable to perform this delete ... + String msg = GraphI18n.unableToDeletePlaceholder.text(readable(request.at()), request.inWorkspace(), getSourceName()); + request.setError(new UnsupportedRequestException(msg)); + } + } + + /** + * {@inheritDoc} + * * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.DeleteBranchRequest) */ @Override @@ -1160,7 +1267,7 @@ // A delete should be executed against any ProxyNode that applies ... FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); - boolean submit = deleteBranch(federatedRequest, projectedNode, workspace, getExecutionContext()); + boolean submit = deleteBranch(federatedRequest, projectedNode, workspace, getExecutionContext(), true); if (submit) { // Submit the requests for processing and then STOP ... submit(federatedRequest); @@ -1174,13 +1281,14 @@ protected boolean deleteBranch( FederatedRequest federatedRequest, ProjectedNode projectedNode, FederatedWorkspace workspace, - ExecutionContext context ) { + ExecutionContext context, + boolean includeParent ) { boolean submit = false; while (projectedNode != null) { if (projectedNode.isProxy()) { ProxyNode proxy = projectedNode.asProxy(); // Is this proxy represent the top level node of the projection? - if (proxy.isTopLevelNode()) { + if (proxy.isTopLevelNode() || !includeParent) { // Then we want to delete everything *underneath* the node, but we don't want to delete // the node itself since it is the node being projected and must exist in order for the // projection to work. @@ -1194,17 +1302,19 @@ submit = true; } else if (projectedNode.isPlaceholder()) { PlaceholderNode placeholder = projectedNode.asPlaceholder(); - // Create a delete for this placeholder, but mark it completed. This is needed to know - // which placeholders were being deleted. - DeleteBranchRequest delete = new DeleteBranchRequest(placeholder.location(), workspace.getName()); - delete.setActualLocationOfNode(placeholder.location()); - federatedRequest.add(delete, true, true, null); + if (includeParent) { + // Create a delete for this placeholder, but mark it completed. This is needed to know + // which placeholders were being deleted. + DeleteBranchRequest delete = new DeleteBranchRequest(placeholder.location(), workspace.getName()); + delete.setActualLocationOfNode(placeholder.location()); + federatedRequest.add(delete, true, true, null); + } // Create and submit a request for each proxy below this placeholder ... // For each child of the placeholder node ... for (ProjectedNode child : placeholder.children()) { while (child != null && child.isProxy()) { // Call recursively ... - submit = deleteBranch(federatedRequest, child.asProxy(), workspace, context); + submit = deleteBranch(federatedRequest, child.asProxy(), workspace, context, true); child = child.next(); } } @@ -1341,57 +1451,111 @@ // Figure out where the 'from' is projected ... ProjectedNode projectedFromNode = project(request.from(), request.inWorkspace(), request, true); if (projectedFromNode == null) return; - ProjectedNode projectedIntoNode = project(request.into(), request.inWorkspace(), request, true); - if (projectedIntoNode == null) return; + ProjectedNode projectedBeforeNode = request.before() != null ? project(request.before(), request.inWorkspace(), request, true) : null; - // Limitation: only able to project the move if the 'from' and 'into' are in the same source & projection ... - while (projectedFromNode != null) { - if (projectedFromNode.isProxy()) { - ProxyNode fromProxy = projectedFromNode.asProxy(); - // Look for a projectedIntoNode that has the same source/projection ... - while (projectedIntoNode != null) { - if (projectedIntoNode.isProxy()) { - // Both are proxies, so compare the projection ... - ProxyNode intoProxy = projectedIntoNode.asProxy(); - if (fromProxy.projection().getSourceName().equals(intoProxy.projection().getSourceName())) break; + boolean sameLocation = true; + if (request.into() != null) { + // Look at where the node is being moved; it must be within the same source/projection ... + ProjectedNode projectedIntoNode = project(request.into(), request.inWorkspace(), request, true); + if (projectedIntoNode == null) return; + // Limitation: only able to project the move if the 'from' and 'into' are in the same source & projection ... + while (projectedFromNode != null) { + if (projectedFromNode.isProxy()) { + ProxyNode fromProxy = projectedFromNode.asProxy(); + // Look for a projectedIntoNode that has the same source/projection ... + while (projectedIntoNode != null) { + if (projectedIntoNode.isProxy()) { + // Both are proxies, so compare the projection ... + ProxyNode intoProxy = projectedIntoNode.asProxy(); + if (fromProxy.projection().getSourceName().equals(intoProxy.projection().getSourceName())) break; + } + projectedIntoNode = projectedIntoNode.next(); } - projectedIntoNode = projectedIntoNode.next(); + if (projectedIntoNode != null) break; } - if (projectedIntoNode != null) break; + projectedFromNode = projectedFromNode.next(); } - projectedFromNode = projectedFromNode.next(); + if (projectedFromNode == null || projectedIntoNode == null) { + // The move is not done within a single source ... + String msg = GraphI18n.moveLimitedToBeWithinSingleSource.text(readable(request.from()), + request.inWorkspace(), + readable(request.into()), + request.inWorkspace(), + getSourceName()); + request.setError(new UnsupportedRequestException(msg)); + return; + } + + ProxyNode fromProxy = projectedFromNode.asProxy(); + ProxyNode intoProxy = projectedIntoNode.asProxy(); + ProxyNode beforeProxy = request.before() != null ? projectedBeforeNode.asProxy() : null; + assert fromProxy.projection().getSourceName().equals(intoProxy.projection().getSourceName()); + sameLocation = fromProxy.isSameLocationAsOriginal() && intoProxy.isSameLocationAsOriginal(); + + // Create the pushed-down request ... + Location beforeProxyLocation = beforeProxy != null ? beforeProxy.location() : null; + MoveBranchRequest pushDown = new MoveBranchRequest(fromProxy.location(), intoProxy.location(), beforeProxyLocation, + intoProxy.workspaceName(), request.desiredName(), + request.conflictBehavior()); + // Create the federated request ... + FederatedRequest federatedRequest = new FederatedRequest(request); + federatedRequest.add(pushDown, sameLocation, false, fromProxy.projection(), intoProxy.projection()); + + // Submit the requests for processing and then STOP ... + submit(federatedRequest); + } else { + ProxyNode fromProxy = projectedFromNode.asProxy(); + ProxyNode beforeProxy = request.before() != null ? projectedBeforeNode.asProxy() : null; + Location beforeProxyLocation = beforeProxy != null ? beforeProxy.location() : null; + MoveBranchRequest pushDown = new MoveBranchRequest(fromProxy.location(), null, beforeProxyLocation, + fromProxy.workspaceName(), request.desiredName(), + request.conflictBehavior()); + // Create the federated request ... + FederatedRequest federatedRequest = new FederatedRequest(request); + federatedRequest.add(pushDown, sameLocation, false, fromProxy.projection()); + + // Submit the requests for processing and then STOP ... + submit(federatedRequest); } - if (projectedFromNode == null || projectedIntoNode == null) { - // The move is not done within a single source ... - String msg = GraphI18n.moveLimitedToBeWithinSingleSource.text(readable(request.from()), - request.inWorkspace(), - readable(request.into()), - request.inWorkspace(), - getSourceName()); - request.setError(new UnsupportedRequestException(msg)); - return; - } + } - ProxyNode fromProxy = projectedFromNode.asProxy(); - ProxyNode intoProxy = projectedIntoNode.asProxy(); - ProxyNode beforeProxy = request.before() != null ? projectedBeforeNode.asProxy() : null; - assert fromProxy.projection().getSourceName().equals(intoProxy.projection().getSourceName()); - boolean sameLocation = fromProxy.isSameLocationAsOriginal() && intoProxy.isSameLocationAsOriginal(); + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.RenameNodeRequest) + */ + @Override + public void process( RenameNodeRequest request ) { + // Figure out where the 'at' is projected ... + ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, true); + if (projectedNode == null) return; - // Create the pushed-down request ... - MoveBranchRequest pushDown = new MoveBranchRequest(fromProxy.location(), intoProxy.location(), beforeProxy.location(), - intoProxy.workspaceName(), request.desiredName(), - request.conflictBehavior()); // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); - federatedRequest.add(pushDown, sameLocation, false, fromProxy.projection(), intoProxy.projection()); - // Submit the requests for processing and then STOP ... - submit(federatedRequest); + // Any non-read request should be submitted to the first ProxyNode ... + while (projectedNode != null) { + if (projectedNode.isProxy()) { + ProxyNode proxy = projectedNode.asProxy(); + // Create and submit a request for the projection ... + RenameNodeRequest pushDownRequest = new RenameNodeRequest(proxy.location(), proxy.workspaceName(), + request.toName()); + federatedRequest.add(pushDownRequest, proxy.isSameLocationAsOriginal(), false, proxy.projection()); + + // Submit the requests for processing and then STOP ... + submit(federatedRequest); + return; + } + assert projectedNode.isPlaceholder(); + projectedNode = projectedNode.next(); + } + // Unable to perform this update ... + String msg = GraphI18n.unableToUpdatePlaceholder.text(readable(request.at()), request.inWorkspace(), getSourceName()); + request.setError(new UnsupportedRequestException(msg)); } /** Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/JoinRequestProcessor.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/JoinRequestProcessor.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/JoinRequestProcessor.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -47,6 +47,7 @@ import org.jboss.dna.graph.request.CacheableRequest; import org.jboss.dna.graph.request.CloneBranchRequest; import org.jboss.dna.graph.request.CloneWorkspaceRequest; +import org.jboss.dna.graph.request.CompositeRequest; import org.jboss.dna.graph.request.CopyBranchRequest; import org.jboss.dna.graph.request.CreateNodeRequest; import org.jboss.dna.graph.request.CreateWorkspaceRequest; @@ -61,6 +62,8 @@ import org.jboss.dna.graph.request.ReadBranchRequest; import org.jboss.dna.graph.request.ReadNodeRequest; import org.jboss.dna.graph.request.ReadPropertyRequest; +import org.jboss.dna.graph.request.RemovePropertyRequest; +import org.jboss.dna.graph.request.RenameNodeRequest; import org.jboss.dna.graph.request.Request; import org.jboss.dna.graph.request.SetPropertyRequest; import org.jboss.dna.graph.request.UpdatePropertiesRequest; @@ -104,14 +107,19 @@ * * @param completedFederatedRequests the collection of {@link FederatedRequest} whose projected requests have already been * processed; may not be null + * @return the exception that occurred while processing these requests (analogous to {@link CompositeRequest#getError()} where + * the composite's requests are these errors), or null if there was none * @see FederatedRepositoryConnection#execute(ExecutionContext, org.jboss.dna.graph.request.Request) */ - public void process( final Iterable<FederatedRequest> completedFederatedRequests ) { + public Throwable process( final Iterable<FederatedRequest> completedFederatedRequests ) { for (FederatedRequest federatedRequest : completedFederatedRequests) { // No need to await for the forked request, since it will be done - this.federatedRequest = federatedRequest; - process(federatedRequest.original()); + process(federatedRequest); + if (federatedRequest.original().hasError()) { + return federatedRequest.original().getError(); + } } + return null; } /** @@ -130,40 +138,8 @@ forked = federatedRequestQueue.take(); if (forked instanceof NoMoreFederatedRequests) return; forked.await(); - - // Determine whether this is a single mirror request ... - Request original = forked.original(); - ProjectedRequest projectedRequest = forked.getFirstProjectedRequest(); - boolean sameLocation = projectedRequest != null && !projectedRequest.hasNext() - && projectedRequest.isSameLocation(); - - // Set the cachable information ... - if (original instanceof CacheableRequest) { - CacheableRequest cacheableOriginal = (CacheableRequest)original; - cacheableOriginal.setCachePolicy(getDefaultCachePolicy()); - while (projectedRequest != null) { - Request requestToSource = projectedRequest.getRequest(); - if (cacheableOriginal != null) { - setCacheableInfo(cacheableOriginal, ((CacheableRequest)requestToSource).getCachePolicy()); - } - projectedRequest = projectedRequest.next(); - } - } - - // Now do the join on this request ... - if (sameLocation) { - Request sourceRequest = forked.getFirstProjectedRequest().getRequest(); - if (sourceRequest.hasError()) { - original.setError(sourceRequest.getError()); - } else if (sourceRequest.isCancelled()) { - original.cancel(); - } - mirrorProcessor.setFederatedRequest(forked); - mirrorProcessor.process(original); - } else { - this.federatedRequest = forked; - process(original); - } + // Now process ... + process(forked); } } catch (InterruptedException e) { // This happens when the federated connector has been told to shutdown now, and it shuts down @@ -180,6 +156,41 @@ } } + protected final void process( FederatedRequest forked ) { + // Determine whether this is a single mirror request ... + Request original = forked.original(); + ProjectedRequest projectedRequest = forked.getFirstProjectedRequest(); + boolean sameLocation = projectedRequest != null && !projectedRequest.hasNext() && projectedRequest.isSameLocation(); + + // Set the cachable information ... + if (original instanceof CacheableRequest) { + CacheableRequest cacheableOriginal = (CacheableRequest)original; + cacheableOriginal.setCachePolicy(getDefaultCachePolicy()); + while (projectedRequest != null) { + Request requestToSource = projectedRequest.getRequest(); + if (cacheableOriginal != null) { + setCacheableInfo(cacheableOriginal, ((CacheableRequest)requestToSource).getCachePolicy()); + } + projectedRequest = projectedRequest.next(); + } + } + + // Now do the join on this request ... + if (sameLocation) { + Request sourceRequest = forked.getFirstProjectedRequest().getRequest(); + if (sourceRequest.hasError()) { + original.setError(sourceRequest.getError()); + } else if (sourceRequest.isCancelled()) { + original.cancel(); + } + mirrorProcessor.setFederatedRequest(forked); + mirrorProcessor.process(original); + } else { + this.federatedRequest = forked; + process(original); + } + } + /** * {@inheritDoc} * @@ -850,6 +861,21 @@ /** * {@inheritDoc} * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.RemovePropertyRequest) + */ + @Override + public void process( RemovePropertyRequest request ) { + ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); + assert !projected.hasNext(); + SetPropertyRequest source = (SetPropertyRequest)projected.getRequest(); + if (checkErrorOrCancel(request, source)) return; + Location sourceLocation = source.getActualLocationOfNode(); + request.setActualLocationOfNode(projectToFederated(request.from(), projected.getProjection(), sourceLocation, request)); + } + + /** + * {@inheritDoc} + * * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.DeleteBranchRequest) */ @Override @@ -885,6 +911,51 @@ /** * {@inheritDoc} * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.DeleteChildrenRequest) + */ + @Override + public void process( DeleteChildrenRequest request ) { + ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); + // Go through the projected requests, and look for the top-most node ... + Location highest = null; + while (projected != null) { + // The projected request should a DeleteChildrenRequest ... + Request sourceRequest = projected.getRequest(); + DeleteChildrenRequest source = (DeleteChildrenRequest)projected.getRequest(); + Location actual = source.getActualLocationOfNode(); + if (checkErrorOrCancel(request, sourceRequest)) return; + if (!projected.isSameLocation() && projected.getProjection() != null) { + actual = projectToFederated(request.at(), projected.getProjection(), actual, request); + } + if (highest == null) highest = actual; + else if (highest.getPath().isDecendantOf(actual.getPath())) highest = actual; + projected = projected.next(); + } + assert highest != null; + request.setActualLocationOfNode(highest); + } + + /** + * {@inheritDoc} + * + * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.RenameNodeRequest) + */ + @Override + public void process( RenameNodeRequest request ) { + ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); + assert !projected.hasNext(); + RenameNodeRequest source = (RenameNodeRequest)projected.getRequest(); + if (checkErrorOrCancel(request, source)) return; + Location locationBefore = source.getActualLocationBefore(); + Location locationAfter = source.getActualLocationBefore(); + locationBefore = projectToFederated(request.at(), projected.getProjection(), locationBefore, request); + locationAfter = projectToFederated(request.at(), projected.getSecondProjection(), locationAfter, request); + request.setActualLocations(locationBefore, locationAfter); + } + + /** + * {@inheritDoc} + * * @see org.jboss.dna.graph.request.processor.RequestProcessor#process(org.jboss.dna.graph.request.CopyBranchRequest) */ @Override @@ -932,7 +1003,9 @@ Location locationBefore = source.getActualLocationBefore(); Location locationAfter = source.getActualLocationBefore(); locationBefore = projectToFederated(request.from(), projected.getProjection(), locationBefore, request); - locationAfter = projectToFederated(request.into(), projected.getSecondProjection(), locationAfter, request); + Projection afterProjection = projected.getSecondProjection(); + if (afterProjection == null) projected.getProjection(); + locationAfter = projectToFederated(request.into(), afterProjection, locationAfter, request); request.setActualLocations(locationBefore, locationAfter); } Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/CompositeRequest.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/CompositeRequest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/CompositeRequest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import net.jcip.annotations.Immutable; import org.jboss.dna.common.util.CheckArg; +import org.jboss.dna.graph.GraphI18n; /** * A request that wraps multiple other requests, allowing multiple requests to be treated as a single request. @@ -258,6 +259,50 @@ } /** + * Check the set of requests for errors. Calling this method will always result in the {@link #getError() existing error} + * being reset to a new error or null if there are no errors. + */ + public void checkForErrors() { + Throwable firstError = null; + LinkedList<Throwable> additionalErrors = null; + for (Request request : requests) { + if (request.hasError()) { + if (firstError == null) { + firstError = request.getError(); + } else { + if (additionalErrors == null) additionalErrors = new LinkedList<Throwable>(); + additionalErrors.add(request.getError()); + } + } + } + if (firstError != null) { + // There is at least one error ... + if (additionalErrors == null) { + // but only one ... + setError(firstError); + return; + } + // More than one ... + additionalErrors.addFirst(firstError); + // Now build a single composite error message ... + StringBuilder str = new StringBuilder(); + for (Throwable error : additionalErrors) { + str.append("\n"); + str.append("\t" + error.getMessage()); + } + String msg = null; + int numberOfErrors = additionalErrors.size(); + int numberOfRequests = size(); + if (numberOfRequests == CompositeRequest.UNKNOWN_NUMBER_OF_REQUESTS) { + msg = GraphI18n.multipleErrorsWhileExecutingManyRequests.text(numberOfErrors, str.toString()); + } else { + msg = GraphI18n.multipleErrorsWhileExecutingRequests.text(numberOfErrors, numberOfRequests, str.toString()); + } + setError(new RequestException(msg)); + } + } + + /** * {@inheritDoc} * * @see java.lang.Object#equals(java.lang.Object) Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidRequestException.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidRequestException.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidRequestException.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -28,7 +28,7 @@ * * @author Randall Hauch */ -public class InvalidRequestException extends RuntimeException { +public class InvalidRequestException extends RequestException { /** */ Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidWorkspaceException.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidWorkspaceException.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidWorkspaceException.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -28,7 +28,7 @@ * * @author Randall Hauch */ -public class InvalidWorkspaceException extends RuntimeException { +public class InvalidWorkspaceException extends RequestException { /** */ @@ -61,7 +61,7 @@ * @param cause */ public InvalidWorkspaceException( String message, - Throwable cause ) { + Throwable cause ) { super(message, cause); } Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/MoveBranchRequest.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/MoveBranchRequest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/MoveBranchRequest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -141,7 +141,7 @@ /** * Get the location defining the parent where the branch is to be placed * - * @return the to location; never null + * @return the to location; or null if the node is being reordered within the same parent */ public Location into() { return into; Copied: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/RequestException.java (from rev 1117, trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/InvalidRequestException.java) =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/RequestException.java (rev 0) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/RequestException.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -0,0 +1,69 @@ +/* + * JBoss DNA (http://www.jboss.org/dna) + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * See the AUTHORS.txt file in the distribution for a full listing of + * individual contributors. + * + * JBoss DNA is free software. Unless otherwise indicated, all code in JBoss DNA + * is licensed to you under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * JBoss DNA is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.dna.graph.request; + +/** + * Specifies that the request was invalid and could not be completed. + * + * @author Randall Hauch + */ +public class RequestException extends RuntimeException { + + /** + */ + private static final long serialVersionUID = 1L; + + /** + * + */ + public RequestException() { + } + + /** + * @param message + */ + public RequestException( String message ) { + super(message); + + } + + /** + * @param cause + */ + public RequestException( Throwable cause ) { + super(cause); + + } + + /** + * @param message + * @param cause + */ + public RequestException( String message, + Throwable cause ) { + super(message, cause); + + } + +} Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/UnsupportedRequestException.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/UnsupportedRequestException.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/UnsupportedRequestException.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -28,7 +28,7 @@ * * @author Randall Hauch */ -public class UnsupportedRequestException extends RuntimeException { +public class UnsupportedRequestException extends RequestException { /** */ Modified: trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/processor/RequestProcessor.java =================================================================== --- trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/processor/RequestProcessor.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/request/processor/RequestProcessor.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -34,7 +34,6 @@ import org.jboss.dna.graph.GraphI18n; import org.jboss.dna.graph.Location; import org.jboss.dna.graph.cache.CachePolicy; -import org.jboss.dna.graph.connector.RepositorySourceException; import org.jboss.dna.graph.observe.Changes; import org.jboss.dna.graph.observe.Observer; import org.jboss.dna.graph.property.DateTime; @@ -283,39 +282,18 @@ */ public void process( CompositeRequest request ) { if (request == null) return; - int numberOfErrors = 0; - List<Throwable> errors = null; + boolean hasErrors = false; // Iterate over the requests in this composite, but only iterate once so that for (Request embedded : request) { assert embedded != null; if (embedded.isCancelled()) return; process(embedded); - if (embedded.hasError()) { - if (numberOfErrors == 0) { - errors = new LinkedList<Throwable>(); - } - assert errors != null; - errors.add(embedded.getError()); - ++numberOfErrors; + if (!hasErrors && embedded.hasError()) { + hasErrors = true; } } - if (numberOfErrors == 0) return; - assert errors != null; - if (numberOfErrors == 1) { - request.setError(errors.get(0)); - } else { - StringBuilder errorString = new StringBuilder(); - for (Throwable error : errors) { - errorString.append("\n"); - errorString.append("\t" + error.getMessage()); - } - String msg = null; - if (request.size() == CompositeRequest.UNKNOWN_NUMBER_OF_REQUESTS) { - msg = GraphI18n.multipleErrorsWhileExecutingManyRequests.text(numberOfErrors, errorString.toString()); - } else { - msg = GraphI18n.multipleErrorsWhileExecutingRequests.text(numberOfErrors, request.size(), errorString.toString()); - } - request.setError(new RepositorySourceException(getSourceName(), msg)); + if (hasErrors) { + request.checkForErrors(); } } @@ -732,6 +710,9 @@ request.setError(update.getError()); } // Set the actual location ... + if (update.getActualLocationOfNode() == null) { + int x = 0; + } request.setActualLocationOfNode(update.getActualLocationOfNode()); } Modified: trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/AbstractFederatedRepositorySourceIntegrationTest.java =================================================================== --- trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/AbstractFederatedRepositorySourceIntegrationTest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/AbstractFederatedRepositorySourceIntegrationTest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -24,6 +24,7 @@ package org.jboss.dna.graph.connector.federation; import static org.hamcrest.core.Is.is; +import static org.hamcrest.core.IsNull.notNullValue; import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; import static org.mockito.Mockito.stub; @@ -38,6 +39,7 @@ import org.jboss.dna.graph.Graph; import org.jboss.dna.graph.Location; import org.jboss.dna.graph.Node; +import org.jboss.dna.graph.Results; import org.jboss.dna.graph.Subgraph; import org.jboss.dna.graph.connector.RepositoryConnection; import org.jboss.dna.graph.connector.RepositoryConnectionFactory; @@ -342,4 +344,15 @@ AbstractConnectorTest.assertEquivalentSubgraphs(fedSubgraph, sourceSubgraph, true, false); } } + + protected void assertReadUsingBatch( String... pathsInFederated ) { + Graph.Batch batch = federated.batch(); + for (String pathInFederated : pathsInFederated) { + batch.read(pathInFederated).and(); + } + Results results = batch.execute(); + for (String pathInFederated : pathsInFederated) { + assertThat(results.getNode(pathInFederated), is(notNullValue())); + } + } } Modified: trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnectionTest.java =================================================================== --- trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnectionTest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositoryConnectionTest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -34,6 +34,7 @@ import org.jboss.dna.graph.request.CompositeRequest; import org.jboss.dna.graph.request.Request; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.mockito.MockitoAnnotations; import org.mockito.MockitoAnnotations.Mock; @@ -92,6 +93,7 @@ assertThat(connection.shouldProcessSynchronously(request), is(true)); } + @Ignore @Test public void shouldProcessCompositeRequestWithMultipleRequestsAsynchronously() { CompositeRequest request = mock(CompositeRequest.class); Modified: trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySourceUsingMirrorAndBranchProjectionsTest.java =================================================================== --- trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySourceUsingMirrorAndBranchProjectionsTest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/FederatedRepositorySourceUsingMirrorAndBranchProjectionsTest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -138,6 +138,11 @@ } @Test + public void shouldPerformBatchOperationsAgainstFederatedSource() { + assertReadUsingBatch("/Aircraft", "/Aircraft/Business", "/Cars"); + } + + @Test public void shouldCreateNodeUnderRootInMirrorSource() { federated.createAt("/Hovercraft").with("prop1", "value1").and(); assertMirrorNode("/Hovercraft"); Modified: trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessorChannelTest.java =================================================================== --- trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessorChannelTest.java 2009-07-20 19:05:35 UTC (rev 1117) +++ trunk/dna-graph/src/test/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessorChannelTest.java 2009-07-21 03:14:59 UTC (rev 1118) @@ -33,6 +33,7 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.jboss.dna.graph.ExecutionContext; @@ -153,6 +154,40 @@ assertThat(iter.hasNext(), is(false)); } + @Test + public void shouldSubmitBlockedRequestsToConnection() throws Exception { + // Start the channel ... + channel.start(executor, context, connectionFactory); + + // Submit the requests to the channel ... + List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); + for (Request request : requests) { + CountDownLatch latch = new CountDownLatch(1); + latches.add(latch); + channel.add(request, latch); + } + + // Mark the channel as done ... + channel.done(); + + // Wait until the channel has completed ... + channel.await(); + + // Verify that all of the latches were decremented ... + for (CountDownLatch latch : latches) { + latch.await(); + assertThat(latch.getCount(), is(0L)); + } + + // Verify that all the requests to the channel were processed ... + Iterator<Request> iter = executedRequests.iterator(); + for (Request expected : requests) { + assertThat(iter.hasNext(), is(true)); + assertThat(iter.next(), is(sameInstance(expected))); + } + assertThat(iter.hasNext(), is(false)); + } + protected static class AddRequestsRunnable implements Runnable { private final Channel channel; private final Iterator<Request> requests;