[dna-commits] DNA SVN: r1519 - trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation.

Author: bcarothers Date: 2010-01-04 10:29:24 -0500 (Mon, 04 Jan 2010) New Revision: 1519 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/JoinRequestProcessor.java Log: DNA-572 FederatedRepositoryConnection Can Freeze When Exception Thrown from RepositoryConnection.execute(...) Applied patch that corrects one possible source of freezes in the federated connector by ensuring that a FederatedRequest is still submitted if the projection fails. 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 2010-01-04 15:26:50 UTC (rev 1518) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/ForkRequestProcessor.java 2010-01-04 15:29:24 UTC (rev 1519) @@ -311,7 +311,6 @@ public void process( VerifyNodeExistsRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -343,7 +342,6 @@ public void process( ReadNodeRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -414,7 +412,6 @@ public void process( ReadAllChildrenRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.of(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -504,7 +501,6 @@ public void process( ReadAllPropertiesRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -538,7 +534,6 @@ public void process( ReadPropertyRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.on(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -578,15 +573,18 @@ public void process( ReadBranchRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.at(), request.inWorkspace(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); - FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); - // And process the branch, creating ReadNodeRequests for each placeholder, and ReadBranchRequests for each proxy node... - if (request.maximumDepth() > 0) { - processBranch(federatedRequest, projectedNode, workspace, request.maximumDepth()); + if (projectedNode != null) { + FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); + + // And process the branch, creating ReadNodeRequests for each placeholder, and ReadBranchRequests for each proxy + // node... + if (request.maximumDepth() > 0) { + processBranch(federatedRequest, projectedNode, workspace, request.maximumDepth()); + } } // Submit the requests for processing ... @@ -645,10 +643,13 @@ public void process( CreateNodeRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.under(), request.inWorkspace(), request, true); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); + if (projectedNode == null) { + submit(federatedRequest); + return; + } // Any non-read request should be submitted to the first ProxyNode ... PlaceholderNode placeholder = null; @@ -724,11 +725,15 @@ 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); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -759,11 +764,15 @@ public void process( UpdatePropertiesRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.on(), request.inWorkspace(), request, true); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -794,11 +803,15 @@ public void process( UpdateValuesRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.on(), request.inWorkspace(), request, true); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -830,11 +843,15 @@ public void process( SetPropertyRequest request ) { // Figure out where this request is projected ... ProjectedNode projectedNode = project(request.on(), request.inWorkspace(), request, true); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -865,11 +882,15 @@ 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); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // A delete should be executed against any ProxyNode that applies ... FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); boolean submit = deleteBranch(federatedRequest, projectedNode, workspace, getExecutionContext(), false); @@ -892,11 +913,15 @@ public void process( DeleteBranchRequest 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); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // A delete should be executed against any ProxyNode that applies ... FederatedWorkspace workspace = getWorkspace(request, request.inWorkspace()); boolean submit = deleteBranch(federatedRequest, projectedNode, workspace, getExecutionContext(), true); @@ -965,9 +990,15 @@ public void process( CopyBranchRequest request ) { // Figure out where the 'from' is projected ... ProjectedNode projectedFromNode = project(request.from(), request.fromWorkspace(), request, false); - if (projectedFromNode == null) return; + if (projectedFromNode == null) { + submit(new FederatedRequest(request)); + return; + } ProjectedNode projectedIntoNode = project(request.into(), request.intoWorkspace(), request, true); - if (projectedIntoNode == null) return; + if (projectedIntoNode == null) { + submit(new FederatedRequest(request)); + return; + } // Limitation: only able to project the copy if the 'from' and 'into' are in the same source & projection ... while (projectedFromNode != null) { @@ -1023,9 +1054,15 @@ public void process( CloneBranchRequest request ) { // Figure out where the 'from' is projected ... ProjectedNode projectedFromNode = project(request.from(), request.fromWorkspace(), request, false); - if (projectedFromNode == null) return; + if (projectedFromNode == null) { + submit(new FederatedRequest(request)); + return; + } ProjectedNode projectedIntoNode = project(request.into(), request.intoWorkspace(), request, true); - if (projectedIntoNode == null) return; + if (projectedIntoNode == null) { + submit(new FederatedRequest(request)); + return; + } // Limitation: only able to project the copy if the 'from' and 'into' are in the same source & projection ... while (projectedFromNode != null) { @@ -1082,7 +1119,10 @@ public void process( MoveBranchRequest request ) { // Figure out where the 'from' is projected ... ProjectedNode projectedFromNode = project(request.from(), request.inWorkspace(), request, true); - if (projectedFromNode == null) return; + if (projectedFromNode == null) { + submit(new FederatedRequest(request)); + return; + } ProjectedNode projectedBeforeNode = request.before() != null ? project(request.before(), request.inWorkspace(), @@ -1164,11 +1204,15 @@ 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 federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -1199,11 +1243,15 @@ public void process( LockBranchRequest 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); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -1234,11 +1282,15 @@ public void process( UnlockBranchRequest 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); + if (projectedNode == null) { + submit(federatedRequest); + return; + } + // Any non-read request should be submitted to the first ProxyNode ... while (projectedNode != null) { if (projectedNode.isProxy()) { @@ -1273,7 +1325,6 @@ // Get the root location ... Location root = Location.create(getExecutionContext().getValueFactories().getPathFactory().createRootPath()); ProjectedNode projectedNode = project(root, workspace.getName(), request, false); - if (projectedNode == null) return; // Create the federated request ... FederatedRequest federatedRequest = new FederatedRequest(request); @@ -1378,5 +1429,4 @@ channel.cancel(mayInterruptIfRunning); } } - } 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 2010-01-04 15:26:50 UTC (rev 1518) +++ trunk/dna-graph/src/main/java/org/jboss/dna/graph/connector/federation/JoinRequestProcessor.java 2010-01-04 15:29:24 UTC (rev 1519) @@ -204,7 +204,7 @@ @Override public void process( VerifyNodeExistsRequest request ) { ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.at(); int numMerged = 0; @@ -258,7 +258,7 @@ Map<Name, Property> properties = request.getPropertiesByName(); Map<Name, Integer> childSnsIndexes = new HashMap<Name, Integer>(); ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.at(); int numMerged = 0; @@ -409,7 +409,7 @@ Path federatedPath = request.of().getPath(); Map<Name, Integer> childSnsIndexes = new HashMap<Name, Integer>(); ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.of(); int numMerged = 0; @@ -506,7 +506,7 @@ public void process( ReadAllPropertiesRequest request ) { Map<Name, Property> properties = request.getPropertiesByName(); ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.at(); int numMerged = 0; @@ -568,7 +568,7 @@ @Override public void process( ReadPropertyRequest request ) { ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.on(); int numMerged = 0; @@ -631,7 +631,7 @@ @Override public void process( ReadBranchRequest request ) { ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + request.setCachePolicy(getDefaultCachePolicy()); Location actualLocation = request.at(); int numMerged = 0; @@ -838,8 +838,8 @@ @Override public void process( CreateNodeRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - Request projectedRequest = projected.getRequest(); + + Request projectedRequest = projected == null ? null : projected.getRequest(); // Check the error first ... if (checkErrorOrCancel(request, projectedRequest)) return; @@ -866,8 +866,8 @@ @Override public void process( UpdatePropertiesRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - UpdatePropertiesRequest source = (UpdatePropertiesRequest)projected.getRequest(); + + UpdatePropertiesRequest source = projected == null ? null : (UpdatePropertiesRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location sourceLocation = source.getActualLocationOfNode(); request.setActualLocationOfNode(projectToFederated(request.on(), projected.getProjection(), sourceLocation, request)); @@ -882,8 +882,8 @@ @Override public void process( SetPropertyRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - SetPropertyRequest source = (SetPropertyRequest)projected.getRequest(); + + SetPropertyRequest source = projected == null ? null : (SetPropertyRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; // Set the actual location and created flags ... Location sourceLocation = source.getActualLocationOfNode(); @@ -899,8 +899,8 @@ @Override public void process( RemovePropertyRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - SetPropertyRequest source = (SetPropertyRequest)projected.getRequest(); + + SetPropertyRequest source = projected == null ? null : (SetPropertyRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location sourceLocation = source.getActualLocationOfNode(); request.setActualLocationOfNode(projectToFederated(request.from(), projected.getProjection(), sourceLocation, request)); @@ -914,6 +914,11 @@ @Override public void process( DeleteBranchRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); + + // Do an initial check to make sure that there was no error on the source that prevented projection + Request projectedSource = projected == null ? null : projected.getRequest(); + if (checkErrorOrCancel(request, projectedSource)) return; + // Go through the projected requests, and look for the top-most node ... Location highest = null; while (projected != null) { @@ -949,6 +954,11 @@ @Override public void process( DeleteChildrenRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); + + // Do an initial check to make sure that there was no error on the source that prevented projection + Request projectedSource = projected == null ? null : projected.getRequest(); + if (checkErrorOrCancel(request, projectedSource)) return; + // Go through the projected requests, and look for the top-most node ... Location highest = null; while (projected != null) { @@ -976,8 +986,8 @@ @Override public void process( RenameNodeRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - RenameNodeRequest source = (RenameNodeRequest)projected.getRequest(); + + RenameNodeRequest source = projected == null ? null : (RenameNodeRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location locationBefore = source.getActualLocationBefore(); Location locationAfter = source.getActualLocationBefore(); @@ -994,8 +1004,8 @@ @Override public void process( CopyBranchRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - CopyBranchRequest source = (CopyBranchRequest)projected.getRequest(); + + CopyBranchRequest source = projected == null ? null : (CopyBranchRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location locationBefore = source.getActualLocationBefore(); Location locationAfter = source.getActualLocationBefore(); @@ -1012,8 +1022,8 @@ @Override public void process( CloneBranchRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - CloneBranchRequest source = (CloneBranchRequest)projected.getRequest(); + + CloneBranchRequest source = projected == null ? null : (CloneBranchRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location locationBefore = source.getActualLocationBefore(); Location locationAfter = source.getActualLocationBefore(); @@ -1037,8 +1047,8 @@ @Override public void process( MoveBranchRequest request ) { ProjectedRequest projected = federatedRequest.getFirstProjectedRequest(); - assert !projected.hasNext(); - MoveBranchRequest source = (MoveBranchRequest)projected.getRequest(); + + MoveBranchRequest source = projected == null ? null : (MoveBranchRequest)projected.getRequest(); if (checkErrorOrCancel(request, source)) return; Location locationBefore = source.getActualLocationBefore(); Location locationAfter = source.getActualLocationBefore(); @@ -1057,7 +1067,7 @@ @Override public void process( VerifyWorkspaceRequest request ) { ProjectedRequest projectedRequest = federatedRequest.getFirstProjectedRequest(); - assert projectedRequest != null; + Location actualLocation = Location.create(getExecutionContext().getValueFactories().getPathFactory().createRootPath()); while (projectedRequest != null) { VerifyNodeExistsRequest readFromSource = (VerifyNodeExistsRequest)projectedRequest.getRequest();