inteligr8/alfresco-community-repo
1
0
Fork 0
mirror of https://github.com/Alfresco/alfresco-community-repo.git synced 2025-08-07 17:49:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ability to cancel replication actions (replication 94)

Browse Source 
Cance ability, which feeds down into the transfer service, and feeds it back up to the action tracking service. Also, refactor of the replication service unit tests to do the transactions itself, which is needed to reliably test async actions


git-svn-id: https://svn.alfresco.com/repos/alfresco-enterprise/alfresco/HEAD/root@21387 c4b6b30b-aa2e-2d43-bbcb-ca4b014f7261
This commit is contained in:
Nick Burch
2010-07-23 16:52:54 +00:00
parent ef0e1f9da9
commit 4425a2b4db
7 changed files with 233 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config/alfresco/replication-services-context.xml
View File

@@ -67,11 +67,11 @@
<value>{http://www.alfresco.org/model/content/1.0}folder</value>
</list>
</property>
<property name="nodeService" ref="NodeService" />
<property name="jobLockService" ref="JobLockService" />
<property name="transferService" ref="TransferService" />
<property name="replicationService" ref="ReplicationService" />
<property name="nodeCrawlerFactory" ref="NodeCrawlerFactory" />
<property name="actionTrackingService" ref="actionTrackingService" />
</bean>

1
source/java/org/alfresco/repo/action/ActionTrackingServiceImplTest.java
View File

@@ -33,7 +33,6 @@ import org.alfresco.repo.action.executer.MoveActionExecuter;
import org.alfresco.repo.cache.EhCacheAdapter;
import org.alfresco.repo.content.MimetypeMap;
import org.alfresco.repo.security.authentication.AuthenticationUtil;
import org.alfresco.repo.transaction.RetryingTransactionHelper;
import org.alfresco.service.cmr.action.Action;
import org.alfresco.service.cmr.action.ActionService;
import org.alfresco.service.cmr.action.ActionStatus;

87
source/java/org/alfresco/repo/replication/ReplicationActionExecutor.java
View File

@@ -23,51 +23,52 @@ import java.util.List;
import java.util.Set;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.action.ActionCancelledException;
import org.alfresco.repo.action.executer.ActionExecuterAbstractBase;
import org.alfresco.repo.lock.JobLockService;
import org.alfresco.repo.lock.LockAcquisitionException;
import org.alfresco.repo.transfer.ChildAssociatedNodeFinder;
import org.alfresco.repo.transfer.ContentClassFilter;
import org.alfresco.service.cmr.action.Action;
import org.alfresco.service.cmr.action.ActionTrackingService;
import org.alfresco.service.cmr.action.ParameterDefinition;
import org.alfresco.service.cmr.replication.ReplicationDefinition;
import org.alfresco.service.cmr.replication.ReplicationService;
import org.alfresco.service.cmr.replication.ReplicationServiceException;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.transfer.NodeCrawler;
import org.alfresco.service.cmr.transfer.NodeCrawlerFactory;
import org.alfresco.service.cmr.transfer.TransferCallback;
import org.alfresco.service.cmr.transfer.TransferCancelledException;
import org.alfresco.service.cmr.transfer.TransferDefinition;
import org.alfresco.service.cmr.transfer.TransferEvent;
import org.alfresco.service.cmr.transfer.TransferEventBegin;
import org.alfresco.service.cmr.transfer.TransferEventEnterState;
import org.alfresco.service.cmr.transfer.TransferService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Nick Burch
* @since 3.4
*/
public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
private NodeService nodeService;
/**
* The logger
*/
private static Log logger = LogFactory.getLog(ReplicationActionExecutor.class);
private JobLockService jobLockService;
private TransferService transferService;
private ReplicationService replicationService;
private NodeCrawlerFactory nodeCrawlerFactory;
private ActionTrackingService actionTrackingService;
/**
* By default, we lock for 30 minutes
*/
private long replicationActionLockDuration = 30*60*1000;
/**
* Injects the NodeService bean.
*
* @param nodeService the NodeService.
*/
public void setNodeService(NodeService nodeService)
{
this.nodeService = nodeService;
}
/**
* Injects the JobLockService bean.
*
@@ -108,9 +109,19 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
this.nodeCrawlerFactory = nodeCrawlerFactory;
}
/**
* Injects the ActionTrackingService bean.
*
* @param actionTrackingService the ActionTrackingService.
*/
public void setActionTrackingService(ActionTrackingService actionTrackingService)
{
this.actionTrackingService = actionTrackingService;
}
@Override
protected void addParameterDefinitions(List<ParameterDefinition> paramList) {
// TODO Is this needed?
// Not used - our definitions hold everything on them
}
/**
@@ -182,8 +193,6 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
try {
toTransfer = expandPayload(replicationDef);
} catch(Exception e) {
// TODO - Record the error
System.err.println(e);
lock.close();
throw new ReplicationServiceException("Error processing payload list", e);
}
@@ -202,14 +211,22 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
lock
);
} catch(Exception e) {
// TODO - Record the error
System.err.println(e);
lock.close();
throw new ReplicationServiceException("Error executing transfer", e);
if(! (e instanceof TransferCancelledException))
{
lock.close();
throw new ReplicationServiceException("Error executing transfer", e);
}
}
// All done
// All done, release our lock
lock.close();
// If we were cancelled, throw the magic exception so
// that this is correctly recorded
if(actionTrackingService.isCancellationRequested(replicationDef))
{
throw new ActionCancelledException(replicationDef);
}
}
/**
@@ -219,6 +236,7 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
protected class ReplicationDefinitionLockExtender implements TransferCallback
{
private ReplicationDefinition replicationDef;
private String transferId;
private String lockToken;
protected ReplicationDefinitionLockExtender(ReplicationDefinition replicationDef)
@@ -233,11 +251,36 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase {
*/
public void processEvent(TransferEvent event)
{
// Extend our lock
refreshLock();
// If it's the enter event, do skip
if(event instanceof TransferEventEnterState)
{
return;
}
// TODO - Check to see if cancel was requested
// TODO - If it was, use TransferService.cancelAsync(transferId)
// If this is a begin event, make a note of the ID
if(event instanceof TransferEventBegin)
{
transferId = ((TransferEventBegin)event).getTransferId();
}
// Has someone tried to cancel us?
if(actionTrackingService.isCancellationRequested(replicationDef))
{
// Tell the transfer service to cancel, if we can
if(transferId != null)
{
transferService.cancelAsync(transferId);
}
else
{
logger.warn("Unable to cancel replication as requested, as transfer has yet to reach a cancellable state");
}
}
}
/**
* Give up our lock on the
* {@link ReplicationDefinition}

159
source/java/org/alfresco/repo/replication/ReplicationServiceIntegrationTest.java
View File

@@ -26,6 +26,10 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.transaction.UserTransaction;
import junit.framework.TestCase;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.lock.JobLockService;
import org.alfresco.repo.model.Repository;
@@ -36,10 +40,13 @@ import org.alfresco.repo.transfer.UnitTestInProcessTransmitterImpl;
import org.alfresco.repo.transfer.UnitTestTransferManifestNodeFactory;
import org.alfresco.repo.transfer.manifest.TransferManifestNodeFactory;
import org.alfresco.service.cmr.action.ActionService;
import org.alfresco.service.cmr.action.ActionStatus;
import org.alfresco.service.cmr.action.ActionTrackingService;
import org.alfresco.service.cmr.replication.ReplicationDefinition;
import org.alfresco.service.cmr.replication.ReplicationService;
import org.alfresco.service.cmr.replication.ReplicationServiceException;
import org.alfresco.service.cmr.repository.ChildAssociationRef;
import org.alfresco.service.cmr.repository.ContentService;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.Path;
@@ -51,22 +58,32 @@ import org.alfresco.service.cmr.transfer.TransferTarget;
import org.alfresco.service.namespace.NamespaceService;
import org.alfresco.service.namespace.QName;
import org.alfresco.service.transaction.TransactionService;
import org.alfresco.util.ApplicationContextHelper;
import org.alfresco.util.BaseAlfrescoSpringTest;
import org.alfresco.util.GUID;
import org.alfresco.util.Pair;
import org.springframework.context.ConfigurableApplicationContext;
/**
* Unit tests for the Replication Service.
* Handles its own transactions, as in a few cases it needs
* to run async actions and know how they'll behave
* @author Nick Burch
*/
@SuppressWarnings("deprecation")
public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
public class ReplicationServiceIntegrationTest extends TestCase
{
private static ConfigurableApplicationContext ctx =
(ConfigurableApplicationContext)ApplicationContextHelper.getApplicationContext();
private ReplicationActionExecutor replicationActionExecutor;
private ReplicationService replicationService;
private TransactionService transactionService;
private TransferService transferService;
private JobLockService jobLockService;
private ActionService actionService;
private NodeService nodeService;
private Repository repositoryHelper;
private ActionTrackingService actionTrackingService;
private NodeRef replicationRoot;
@@ -88,28 +105,25 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
private final String TRANSFER_TARGET = "TestTransferTarget";
public ReplicationServiceIntegrationTest()
{
super();
preventTransaction();
}
@Override
protected void onSetUp() throws Exception
protected void setUp() throws Exception
{
super.onSetUp();
replicationActionExecutor = (ReplicationActionExecutor) this.applicationContext.getBean("replicationActionExecutor");
replicationService = (ReplicationService) this.applicationContext.getBean("ReplicationService");
transactionService = (TransactionService) this.applicationContext.getBean("transactionService");
transferService = (TransferService) this.applicationContext.getBean("TransferService");
jobLockService = (JobLockService) this.applicationContext.getBean("JobLockService");
actionService = (ActionService) this.applicationContext.getBean("ActionService");
nodeService = (NodeService) this.applicationContext.getBean("NodeService");
repositoryHelper = (Repository) this.applicationContext.getBean("repositoryHelper");
replicationActionExecutor = (ReplicationActionExecutor) ctx.getBean("replicationActionExecutor");
replicationService = (ReplicationService) ctx.getBean("replicationService");
transactionService = (TransactionService) ctx.getBean("transactionService");
transferService = (TransferService) ctx.getBean("transferService");
jobLockService = (JobLockService) ctx.getBean("jobLockService");
actionService = (ActionService) ctx.getBean("actionService");
nodeService = (NodeService) ctx.getBean("nodeService");
repositoryHelper = (Repository) ctx.getBean("repositoryHelper");
actionTrackingService = (ActionTrackingService) ctx.getBean("actionTrackingService");
// Set the current security context as admin
AuthenticationUtil.setFullyAuthenticatedUser(AuthenticationUtil.getAdminUserName());
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
// Zap any existing replication entries
replicationRoot = ReplicationDefinitionPersisterImpl.REPLICATION_ACTION_ROOT_NODE_REF;
for(ChildAssociationRef child : nodeService.getChildAssocs(replicationRoot)) {
@@ -136,12 +150,16 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
// Tell the transfer service not to use HTTP
makeTransferServiceLocal();
// Finish setup
txn.commit();
}
@Override
protected void onTearDown() throws Exception {
super.onTearDown();
protected void tearDown() throws Exception {
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
// Zap our test folders
if(folder1 != null) {
nodeService.deleteNode(folder1);
@@ -153,10 +171,16 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
nodeService.deleteNode(destinationFolder);
}
txn.commit();
txn = transactionService.getUserTransaction();
txn.begin();
// Zap our test transfer target
try {
transferService.deleteTransferTarget(TRANSFER_TARGET);
} catch(TransferException e) {}
txn.commit();
}
@@ -438,9 +462,10 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
rd.getPayload().add( folder1 );
// Will execute without error
startNewTransaction();
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
actionService.executeAction(rd, replicationRoot);
endTransaction();
txn.commit();
// Now with one that's in the repo
@@ -453,9 +478,10 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
rd2 = replicationService.loadReplicationDefinition(ACTION_NAME2);
// Again no errors
startNewTransaction();
txn = transactionService.getUserTransaction();
txn.begin();
actionService.executeAction(rd2, replicationRoot);
endTransaction();
txn.commit();
}
/**
@@ -482,9 +508,11 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
1,
1
);
startNewTransaction();
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
actionService.executeAction(rd, replicationRoot);
endTransaction();
txn.commit();
long end = System.currentTimeMillis();
assertTrue(
@@ -494,12 +522,75 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
);
}
/**
* Check that cancelling works.
* Does this by taking a lock on the job, cancelling,
* releasing and seeing it abort.
*/
public void testReplicationExectionCancelling() throws Exception
{
// We need the test transfer target for this test
makeTransferTarget();
// Create a task
ReplicationDefinition rd = replicationService.createReplicationDefinition(ACTION_NAME, "Test");
rd.setTargetName(TRANSFER_TARGET);
rd.getPayload().add(folder1);
rd.getPayload().add(folder2);
// Get the lock for 2 seconds
String token = jobLockService.getLock(
rd.getReplicationName(),
2 * 1000,
1,
1
);
// Request it be run async
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
actionService.executeAction(rd, replicationRoot, false, true);
assertEquals(ActionStatus.Pending, rd.getExecutionStatus());
txn.commit();
// Let it get going, will be waiting for the lock
// having registered with the action tracking service
Thread.sleep(500);
assertEquals(ActionStatus.Running, rd.getExecutionStatus());
// Now request the cancel
actionTrackingService.requestActionCancellation(rd);
// Release our lock, should allow the replication task
// to get going and spot the cancel
jobLockService.releaseLock(token, rd.getReplicationName());
// Let the main replication task run to cancelled/completed
// This can take quite some time though...
for(int i=0; i<10; i++) {
if(rd.getExecutionStatus() == ActionStatus.Running) {
Thread.sleep(1000);
} else {
// It has finished running, check it
break;
}
}
// Ensure it was cancelled
assertEquals(null, rd.getExecutionFailureMessage());
assertEquals(ActionStatus.Cancelled, rd.getExecutionStatus());
}
/**
* Test that when we execute a replication task, the
* right stuff ends up being moved for us
*/
public void testExecutionResult() throws Exception
{
UserTransaction txn = transactionService.getUserTransaction();
txn.begin();
// Destination is empty
assertEquals(0, nodeService.getChildAssocs(destinationFolder).size());
@@ -509,6 +600,8 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
// Put in Folder 2, so we can send Folder 2a
String folder2Name = (String)nodeService.getProperties(folder2).get(ContentModel.PROP_NAME);
NodeRef folderT2 = makeNode(destinationFolder, ContentModel.TYPE_FOLDER, folder2Name);
txn.commit();
// Run a transfer
ReplicationDefinition rd = replicationService.createReplicationDefinition(ACTION_NAME, "Test");
@@ -516,9 +609,10 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
rd.getPayload().add(folder1);
rd.getPayload().add(folder2a);
startNewTransaction();
txn = transactionService.getUserTransaction();
txn.begin();
actionService.executeAction(rd, replicationRoot);
endTransaction();
txn.commit();
// Correct things have turned up
assertEquals(2, nodeService.getChildAssocs(destinationFolder).size());
@@ -704,9 +798,10 @@ public class ReplicationServiceIntegrationTest extends BaseAlfrescoSpringTest
}
private void makeTransferServiceLocal() {
TransferReceiver receiver = (TransferReceiver)this.applicationContext.getBean("transferReceiver");
TransferManifestNodeFactory transferManifestNodeFactory = (TransferManifestNodeFactory)this.applicationContext.getBean("transferManifestNodeFactory");
TransferServiceImpl transferServiceImpl = (TransferServiceImpl) this.applicationContext.getBean("transferService");
TransferReceiver receiver = (TransferReceiver)ctx.getBean("transferReceiver");
TransferManifestNodeFactory transferManifestNodeFactory = (TransferManifestNodeFactory)ctx.getBean("transferManifestNodeFactory");
TransferServiceImpl transferServiceImpl = (TransferServiceImpl) ctx.getBean("transferService");
ContentService contentService = (ContentService) ctx.getBean("contentService");
TransferTransmitter transmitter =
new UnitTestInProcessTransmitterImpl(receiver, contentService, transactionService);

4
source/java/org/alfresco/repo/transfer/TransferServiceImpl.java
View File

@@ -70,6 +70,7 @@ import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.search.ResultSet;
import org.alfresco.service.cmr.search.SearchService;
import org.alfresco.service.cmr.transfer.TransferCallback;
import org.alfresco.service.cmr.transfer.TransferCancelledException;
import org.alfresco.service.cmr.transfer.TransferDefinition;
import org.alfresco.service.cmr.transfer.TransferEvent;
import org.alfresco.service.cmr.transfer.TransferException;
@@ -94,7 +95,6 @@ public class TransferServiceImpl implements TransferService
private static final String MSG_NO_TARGET = "transfer_service.unable_to_find_transfer_target";
private static final String MSG_ERR_TRANSFER_ASYNC = "transfer_service.unable_to_transfer_async";
private static final String MSG_TARGET_EXISTS = "transfer_service.target_exists";
private static final String MSG_CANCELLED = "transfer_service.cancelled";
private static final String MSG_NO_NODES = "transfer_service.no_nodes";
/**
@@ -908,7 +908,7 @@ public class TransferServiceImpl implements TransferService
{
if(status.cancelMe)
{
throw new TransferException(MSG_CANCELLED);
throw new TransferCancelledException();
}
}
}

4
source/java/org/alfresco/service/cmr/replication/ReplicationDefinition.java
View File

@@ -21,7 +21,7 @@ package org.alfresco.service.cmr.replication;
import java.io.Serializable;
import java.util.List;
import org.alfresco.service.cmr.action.Action;
import org.alfresco.service.cmr.action.CancellableAction;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.namespace.QName;
@@ -35,7 +35,7 @@ import org.alfresco.service.namespace.QName;
*
* @author Nick Burch
*/
public interface ReplicationDefinition extends Action, Serializable {
public interface ReplicationDefinition extends CancellableAction, Serializable {
/**
* @return the name which uniquely identifies this replication definition.
*/

36
source/java/org/alfresco/service/cmr/transfer/TransferCancelledException.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright (C) 2005-2010 Alfresco Software Limited.
*
* This file is part of Alfresco
*
* Alfresco is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Alfresco 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 Alfresco. If not, see <http://www.gnu.org/licenses/>.
*/
package org.alfresco.service.cmr.transfer;
/**
* Transfer service exception class for when the
* transfer was halted through a cancel
*
* @author Mark Rogers
*/
public class TransferCancelledException extends TransferException
{
private static final String MSG_CANCELLED = "transfer_service.cancelled";
private static final long serialVersionUID = -1644569346701052090L;
public TransferCancelledException()
{
super(MSG_CANCELLED);
}
}