diff --git a/config/alfresco/messages/transfer-service.properties b/config/alfresco/messages/transfer-service.properties index 0f05d3408b..2e750f539b 100644 --- a/config/alfresco/messages/transfer-service.properties +++ b/config/alfresco/messages/transfer-service.properties @@ -32,6 +32,10 @@ transfer_service.receiver.transfer_not_found=Failed to find any record of reques transfer_service.receiver.transfer_cancelled=Transfer has been cancelled: {0} transfer_service.no_encoding=Unable to deserialize value, no transformation for encoding {0} transfer_service.unable_to_deserialise=Unable to deserialize value +transfer_service.receiver.lock_timed_out=Transfer lock timed out transferId: {0} +transfer_service.receiver.lock_not_found=Transfer lock not found +transfer_service.receiver.error_start=Unable to start a transfer +transfer_service.receiver.error_generating_requisite="Unable to generate transfer requisite transfer_service.missing_endpoint_path=An endpoint path has not been specified for transfer target: {0} transfer_service.missing_endpoint_protocol=An endpoint protocol has not been specified for transfer target: {0} diff --git a/config/alfresco/repository.properties b/config/alfresco/repository.properties index 756ff58662..7560aa2432 100644 --- a/config/alfresco/repository.properties +++ b/config/alfresco/repository.properties @@ -500,7 +500,6 @@ subsystems.test.simpleProp3=Global Default3 deployment.service.numberOfSendingThreads=5 deployment.service.corePoolSize=2 deployment.service.maximumPoolSize=3 - # How long to wait in mS before refreshing a target lock - detects shutdown servers deployment.service.targetLockRefreshTime=60000 # How long to wait in mS from the last communication before deciding that deployment has failed, possibly @@ -510,3 +509,18 @@ deployment.service.targetLockTimeout=3600000 # Transfer Service transferservice.receiver.enabled=true transferservice.receiver.stagingDir=${java.io.tmpdir}/alfresco-transfer-staging +# +# How long to wait in mS before refreshing a transfer lock - detects shutdown servers +# Default 1 minute. +transferservice.receiver.lockRefreshTime=60000 +# +# How many times to attempt retry the transfer lock +transferservice.receiver.lockRetryCount=3 +# How long to wait, in mS, before retrying the transfer lock +transferservice.receiver.lockRetryWait=100 +# +# How long to wait, in mS, since the last contact with from the client before +# timing out a transfer. Needs to be long enough to cope with network delays and "thinking +# time" for both source and destination. Default 5 minutes. +transferservice.receiver.lockTimeOut=300000 + diff --git a/config/alfresco/transfer-service-context.xml b/config/alfresco/transfer-service-context.xml index 496e687a4a..e4faa51677 100644 --- a/config/alfresco/transfer-service-context.xml +++ b/config/alfresco/transfer-service-context.xml @@ -79,6 +79,7 @@ + /${spaces.company_home.childname}/${spaces.dictionary.childname}/${spaces.transfers.childname} @@ -92,6 +93,18 @@ ${transferservice.receiver.stagingDir} + + ${transferservice.receiver.lockRefreshTime} + + + ${transferservice.receiver.lockRetryCount} + + + ${transferservice.receiver.lockRetryWait} + + + ${transferservice.receiver.lockTimeOut} + diff --git a/source/java/org/alfresco/repo/replication/ReplicationActionExecutor.java b/source/java/org/alfresco/repo/replication/ReplicationActionExecutor.java index c5302e182f..7c478c45f0 100644 --- a/source/java/org/alfresco/repo/replication/ReplicationActionExecutor.java +++ b/source/java/org/alfresco/repo/replication/ReplicationActionExecutor.java @@ -75,9 +75,10 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { private ReplicationParams replicationParams; /** - * By default, we lock for 30 minutes + * By default, we lock for a minute, so if this server is shutdown another can take over a + * minute later. */ - private long replicationActionLockDuration = 30*60*1000; + private long replicationActionLockDuration = 60*1000; /** * Injects the NodeService bean. @@ -260,9 +261,11 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { // Turn our payload list of root nodes into something that // the transfer service can work with Set toTransfer; - try { + try + { toTransfer = expandPayload(replicationDef); - } catch(Exception e) { + } + catch(Exception e) { lock.close(); throw new ReplicationServiceException("Error processing payload list - " + e.getMessage(), e); } @@ -346,17 +349,21 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { * A {@link TransferCallback} which periodically renews the * lock held against a {@link ReplicationDefinition} */ - protected class ReplicationDefinitionLockExtender implements TransferCallback + protected class ReplicationDefinitionLockExtender + implements TransferCallback, JobLockService.JobLockRefreshCallback + { private ReplicationDefinition replicationDef; private String transferId; private String lockToken; + private boolean active; protected ReplicationDefinitionLockExtender(ReplicationDefinition replicationDef) { this.replicationDef = replicationDef; acquireLock(); } + /** * No matter what the event is, refresh * our lock on the {@link ReplicationDefinition}, and @@ -364,35 +371,19 @@ 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; - } + // If it's the enter event, do skip + if(event instanceof TransferEventEnterState) + { + return; + } - // 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); - logger.debug("Replication cancel was requested for " + replicationDef.getReplicationQName()); - } - else - { - logger.warn("Unable to cancel replication as requested, as transfer has yet to reach a cancellable state"); - } - } + // If this is a begin event, make a note of the ID + if(event instanceof TransferEventBegin) + { + transferId = ((TransferEventBegin)event).getTransferId(); + } + + checkCancel(); } /** @@ -401,21 +392,20 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { */ public void close() { - releaseLock(); + releaseLock(); } /** * Get a lock on the job. * Tries every 5 seconds for 30 seconds, then - * every 30 seconds until 3 times the lock - * duration. + * every 30 seconds for half an hour. + * + * @throws LockAcquisitionException */ private void acquireLock() { - long retryTime = 30*1000; - int retries = (int)(replicationActionLockDuration * 3 / retryTime); - - try { + try + { // Quick try lockToken = jobLockService.getLock( replicationDef.getReplicationQName(), @@ -423,13 +413,38 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { 5 * 1000, // Every 5 seconds 6 // 6 times = wait up to 30 seconds ); - } catch(LockAcquisitionException e) { + + active = true; + + /** + * Got the lock - now register the refresh callback which will keep the + * lock alive + */ + jobLockService.refreshLock( + lockToken, + replicationDef.getReplicationQName(), + replicationActionLockDuration, + this + ); + + if(logger.isDebugEnabled()) + { + logger.debug("lock aquired:" + replicationDef.getReplicationQName() ); + } + } + catch(LockAcquisitionException e) + { + long retryTime = 30*1000; + int retries = (int)(60); + logger.debug( "Unable to get the replication job lock on " + replicationDef.getReplicationQName() + ", retrying every " + (int)(retryTime/1000) + " seconds" ); + active = true; + // Long try - every 30 seconds lockToken = jobLockService.getLock( replicationDef.getReplicationQName(), @@ -437,22 +452,80 @@ public class ReplicationActionExecutor extends ActionExecuterAbstractBase { retryTime, retries ); + + /** + * Got the lock - now register the refresh callback which will keep the + * lock alive + */ + jobLockService.refreshLock( + lockToken, + replicationDef.getReplicationQName(), + replicationActionLockDuration, + this + ); + + if(logger.isDebugEnabled()) + { + logger.debug("lock aquired (from long timeout):" + replicationDef.getReplicationQName() ); + } } } - private void refreshLock() - { - jobLockService.refreshLock( - lockToken, - replicationDef.getReplicationQName(), - replicationActionLockDuration - ); - } + private void releaseLock() { - jobLockService.releaseLock( - lockToken, - replicationDef.getReplicationQName() - ); + if(active) + { + if(logger.isDebugEnabled()) + { + logger.debug("about to release lock:" + replicationDef.getReplicationQName()); + } + jobLockService.releaseLock( + lockToken, + replicationDef.getReplicationQName()); + active=false; + } + } + + private void checkCancel() + { + // 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); + logger.debug("Replication cancel was requested for " + replicationDef.getReplicationQName()); + } + else + { + logger.warn("Unable to cancel replication as requested, as transfer has yet to reach a cancellable state"); + } + } + } + + /** + * Job Lock Refresh + * @return + */ + @Override + public boolean isActive() + { + if(logger.isDebugEnabled()) + { + logger.debug("lock callback isActive:" + active + ", " + replicationDef.getReplicationQName()); + } + return active; + } + + /** + * Job Lock Service has released us. + */ + @Override + public void lockReleased() + { + logger.debug("lock released:" + replicationDef.getReplicationQName()); + // nothing to do } } } diff --git a/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImpl.java b/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImpl.java index dc8e3e4262..c90ec6cd19 100644 --- a/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImpl.java +++ b/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImpl.java @@ -45,6 +45,8 @@ import org.alfresco.repo.copy.CopyBehaviourCallback; import org.alfresco.repo.copy.CopyDetails; import org.alfresco.repo.copy.CopyServicePolicies; import org.alfresco.repo.copy.DefaultCopyBehaviourCallback; +import org.alfresco.repo.lock.JobLockService; +import org.alfresco.repo.lock.LockAcquisitionException; import org.alfresco.repo.node.NodeServicePolicies; import org.alfresco.repo.policy.BehaviourFilter; import org.alfresco.repo.policy.ClassPolicyDelegate; @@ -64,7 +66,7 @@ import org.alfresco.repo.transfer.requisite.XMLTransferRequsiteWriter; import org.alfresco.service.cmr.action.Action; import org.alfresco.service.cmr.action.ActionService; import org.alfresco.service.cmr.repository.ChildAssociationRef; -import org.alfresco.service.cmr.repository.DuplicateChildNodeNameException; +import org.alfresco.service.cmr.repository.ContentWriter; import org.alfresco.service.cmr.repository.NodeRef; import org.alfresco.service.cmr.repository.NodeService; import org.alfresco.service.cmr.repository.StoreRef; @@ -90,8 +92,17 @@ import org.apache.commons.logging.LogFactory; import org.springframework.util.FileCopyUtils; /** + * The Repo Transfer Receiver is the "Back-End" for transfer subsystem. + *

+ * Provides the implementation of the transfer commands on the destination repository. + *

+ * Provides callback handlers for Aliens and Transferred Aspects. + *

+ * Calls transfer policies. + *

+ * Co-ordinates locking and logging as the transfer progresses. + * * @author brian - * */ public class RepoTransferReceiverImpl implements TransferReceiver, NodeServicePolicies.OnCreateChildAssociationPolicy, @@ -99,7 +110,6 @@ public class RepoTransferReceiverImpl implements TransferReceiver, NodeServicePolicies.OnRestoreNodePolicy, NodeServicePolicies.OnMoveNodePolicy, ContentServicePolicies.OnContentUpdatePolicy - { /** * This embedded class is used to push requests for asynchronous commits onto a different thread @@ -144,20 +154,20 @@ public class RepoTransferReceiverImpl implements TransferReceiver, private final static Log log = LogFactory.getLog(RepoTransferReceiverImpl.class); private static final String MSG_FAILED_TO_CREATE_STAGING_FOLDER = "transfer_service.receiver.failed_to_create_staging_folder"; - private static final String MSG_TRANSFER_LOCK_FOLDER_NOT_FOUND = "transfer_service.receiver.lock_folder_not_found"; + private static final String MSG_ERROR_WHILE_STARTING = "transfer_service.receiver.error_start"; private static final String MSG_TRANSFER_TEMP_FOLDER_NOT_FOUND = "transfer_service.receiver.temp_folder_not_found"; private static final String MSG_TRANSFER_LOCK_UNAVAILABLE = "transfer_service.receiver.lock_unavailable"; private static final String MSG_INBOUND_TRANSFER_FOLDER_NOT_FOUND = "transfer_service.receiver.record_folder_not_found"; - private static final String MSG_NOT_LOCK_OWNER = "transfer_service.receiver.not_lock_owner"; + private static final String MSG_ERROR_WHILE_ENDING_TRANSFER = "transfer_service.receiver.error_ending_transfer"; private static final String MSG_ERROR_WHILE_STAGING_SNAPSHOT = "transfer_service.receiver.error_staging_snapshot"; private static final String MSG_ERROR_WHILE_STAGING_CONTENT = "transfer_service.receiver.error_staging_content"; private static final String MSG_NO_SNAPSHOT_RECEIVED = "transfer_service.receiver.no_snapshot_received"; private static final String MSG_ERROR_WHILE_COMMITTING_TRANSFER = "transfer_service.receiver.error_committing_transfer"; - private static final String MSG_ERROR_WHILE_GENERATING_REQUISITE = "transfer_service.receiver.error_generating_requsite"; + private static final String MSG_ERROR_WHILE_GENERATING_REQUISITE = "transfer_service.receiver.error_generating_requisite"; + private static final String MSG_LOCK_TIMED_OUT = "transfer_service.receiver.lock_timed_out"; + private static final String MSG_LOCK_NOT_FOUND = "transfer_service.receiver.lock_not_found"; - private static final String LOCK_FILE_NAME = ".lock"; - private static final QName LOCK_QNAME = QName.createQName(NamespaceService.APP_MODEL_1_0_URI, LOCK_FILE_NAME); private static final String SNAPSHOT_FILE_NAME = "snapshot.xml"; private NodeService nodeService; @@ -176,17 +186,50 @@ public class RepoTransferReceiverImpl implements TransferReceiver, private PolicyComponent policyComponent; private DescriptorService descriptorService; private AlienProcessor alienProcessor; + private JobLockService jobLockService; - //private String localRepositoryId = descriptorService.getCurrentRepositoryDescriptor().getId(); - - private Map transferLockFolderMap = new ConcurrentHashMap(); + /** + * Where the temporary files are stored. Tenant Domain Name, NodeRef + */ private Map transferTempFolderMap = new ConcurrentHashMap(); + + /** + * Where the destination side transfer report is generated. Tenant Domain Name, NodeRef + */ private Map inboundTransferRecordsFolderMap = new ConcurrentHashMap(); private ClassPolicyDelegate beforeStartInboundTransferDelegate; private ClassPolicyDelegate onStartInboundTransferDelegate; private ClassPolicyDelegate onEndInboundTransferDelegate; + /** + * Locks for the transfers in progress + *

+ * TransferId, Lock + */ + private Map locks = new ConcurrentHashMap(); + + /** + * How many mS before refreshing the lock? + */ + private long lockRefreshTime = 60000; + + /** + * How many times to retry to obtain the lock + */ + private int lockRetryCount = 2; + + /** + * How long to wait between retries + */ + private long lockRetryWait = 100; + + /** + * How long in mS to keep the lock before giving up and ending the transfer, + * possibly the client has terminated? + */ + private long lockTimeOut = 3600000; + public void init() { PropertyCheck.mandatory(this, "nodeService", nodeService); @@ -202,6 +245,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver, PropertyCheck.mandatory(this, "policyComponent", policyComponent); PropertyCheck.mandatory(this, "descriptorService", descriptorService); PropertyCheck.mandatory(this, "alienProcessor", alienProcessor); + PropertyCheck.mandatory(this, "jobLockService", getJobLockService()); beforeStartInboundTransferDelegate = policyComponent.registerClassPolicy(TransferServicePolicies.BeforeStartInboundTransferPolicy.class); onStartInboundTransferDelegate = policyComponent.registerClassPolicy(TransferServicePolicies.OnStartInboundTransferPolicy.class); @@ -301,31 +345,6 @@ public class RepoTransferReceiverImpl implements TransferReceiver, } - private NodeRef getLockFolder() - { - String tenantDomain = tenantService.getUserDomain(AuthenticationUtil.getRunAsUser()); - NodeRef transferLockFolder = transferLockFolderMap.get(tenantDomain); - - // Have we already resolved the node that is the parent of the lock node? - // If not then do so. - if (transferLockFolder == null) - { - ResultSet rs = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, - transferLockFolderPath); - if (rs.length() > 0) - { - transferLockFolder = rs.getNodeRef(0); - transferLockFolderMap.put(tenantDomain, transferLockFolder); - } - else - { - throw new TransferException(MSG_TRANSFER_LOCK_FOLDER_NOT_FOUND, new Object[] { transferLockFolderPath }); - } - } - return transferLockFolder; - - } - public NodeRef getTempFolder(String transferId) { String tenantDomain = tenantService.getUserDomain(AuthenticationUtil.getRunAsUser()); @@ -381,16 +400,36 @@ public class RepoTransferReceiverImpl implements TransferReceiver, */ public String start() { - final NodeRef lockFolder = getLockFolder(); - String transferId = null; - - RetryingTransactionHelper txHelper = transactionService.getRetryingTransactionHelper(); + log.debug("Start transfer"); + + /** + * First get the transfer lock for this domain + */ + String tenantDomain = tenantService.getUserDomain(AuthenticationUtil.getRunAsUser()); + String lockStr = tenantDomain.isEmpty() ? "transfer.server.default" : "transfer.server.tenant." + tenantDomain; + QName lockQName = QName.createQName(TransferModel.TRANSFER_MODEL_1_0_URI, lockStr); + Lock lock = new Lock(lockQName); + try { - transferId = txHelper.doInTransaction( + lock.makeLock(); + + /** + * Transfer Lock held if we get this far + */ + String transferId = null; + + try + { + /** + * Now create a transfer record and use its NodeRef as the transfer id + */ + RetryingTransactionHelper txHelper = transactionService.getRetryingTransactionHelper(); + + transferId = txHelper.doInTransaction( new RetryingTransactionHelper.RetryingTransactionCallback() - { - public String execute() throws Throwable + { + public String execute() throws Throwable { TransferServicePolicies.BeforeStartInboundTransferPolicy beforeStartPolicy = beforeStartInboundTransferDelegate.get(TransferModel.TYPE_TRANSFER_RECORD); @@ -399,24 +438,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver, final NodeRef relatedTransferRecord = createTransferRecord(); String transferId = relatedTransferRecord.toString(); getTempFolder(transferId); - - Map props = new HashMap(); - props.put(ContentModel.PROP_NAME, LOCK_FILE_NAME); - props.put(TransferModel.PROP_TRANSFER_ID, transferId); - - if (log.isInfoEnabled()) - { - log.info("Creating transfer lock associated with this transfer record: " - + relatedTransferRecord); - } - - ChildAssociationRef assoc = nodeService.createNode(lockFolder, ContentModel.ASSOC_CONTAINS, - LOCK_QNAME, TransferModel.TYPE_TRANSFER_LOCK, props); - - if (log.isInfoEnabled()) - { - log.info("Transfer lock created as node " + assoc.getChildRef()); - } + getStagingFolder(transferId); TransferServicePolicies.OnStartInboundTransferPolicy onStartPolicy = onStartInboundTransferDelegate.get(TransferModel.TYPE_TRANSFER_RECORD); @@ -425,15 +447,31 @@ public class RepoTransferReceiverImpl implements TransferReceiver, return transferId; } }, false, true); + } + catch (Exception e) + { + log.debug("Exception while staring transfer", e); + log.debug("releasing lock - we never created the transfer id"); + lock.releaseLock(); + throw new TransferException(MSG_ERROR_WHILE_STARTING, e); + } + + /** + * Here if we have begun a transfer and have a valid transfer id + */ + lock.transferId = transferId; + locks.put(transferId, lock); + log.info("transfer started:" + transferId); + lock.enableLockTimeout(); + return transferId; + } - catch (DuplicateChildNodeNameException ex) + catch (LockAcquisitionException lae) { - log.debug("lock is already taken"); + log.debug("transfer lock is already taken", lae); // lock is already taken. throw new TransferException(MSG_TRANSFER_LOCK_UNAVAILABLE); } - getStagingFolder(transferId); - return transferId; } /** @@ -481,6 +519,61 @@ public class RepoTransferReceiverImpl implements TransferReceiver, return assoc.getChildRef(); } + + /** + * Timeout a transfer. Called after the lock has been released via a timeout. + * + * This is the last chance to clean up. + * + * @param transferId + */ + private void timeout(final String transferId) + { + log.info("Inbound Transfer has timed out transferId:" + transferId); + /* + * There is no transaction or authentication context in this method since it is called via a + * timer thread. + */ + final RetryingTransactionCallback timeoutCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + TransferProgress progress = getProgressMonitor().getProgress(transferId); + + if (progress.getStatus().equals(TransferProgress.Status.PRE_COMMIT)) + { + log.warn("Inbound Transfer Lock Timeout - transferId:" + transferId); + /** + * Did not get out of PRE_COMMIT. The client has probably "gone away" after calling + * "start", but before calling commit, cancel or error. + */ + locks.remove(transferId); + removeTempFolders(transferId); + Object[] msgParams = { transferId }; + getProgressMonitor().logException(transferId, "transfer timeout", new TransferException(MSG_LOCK_TIMED_OUT, msgParams)); + getProgressMonitor().updateStatus(transferId, TransferProgress.Status.ERROR); + } + else + { + // We got beyond PRE_COMMIT, therefore leave the clean up to either + // commit, cancel or error command, since there may still be "in-flight" + // transfer in another thread. Although why, in that case, are we here? + log.warn("Inbound Transfer Lock Timeout - already past PRE-COMMIT - do no cleanup transferId:" + transferId); + } + return null; + } + }; + + AuthenticationUtil.runAs(new AuthenticationUtil.RunAsWork() + { + public String doWork() throws Exception + { + transactionService.getRetryingTransactionHelper().doInTransaction(timeoutCB, false, true); + return null; + } + }, AuthenticationUtil.getSystemUserName()); + } /* * (non-Javadoc) @@ -500,57 +593,16 @@ public class RepoTransferReceiverImpl implements TransferReceiver, try { - // We remove the lock node in a separate transaction, since it was created in a separate transaction - transactionService.getRetryingTransactionHelper().doInTransaction( - new RetryingTransactionHelper.RetryingTransactionCallback() - { - public NodeRef execute() throws Throwable - { - // Find the lock node - NodeRef lockId = getLockNode(); - if (lockId != null) - { - if (!testLockedTransfer(lockId, transferId)) - { - throw new TransferException(MSG_NOT_LOCK_OWNER, new Object[] { transferId }); - } - // Delete the lock node. - log.debug("Deleting lock node :" + lockId); - nodeService.deleteNode(lockId); - log.debug("Lock deleted :" + lockId); - } - return null; - } - }, false, true); - - NodeRef tempStoreNode = null; - try + Lock lock = locks.get(transferId); + if(lock != null) { - log.debug("Deleting temporary store node..."); - tempStoreNode = getTempFolder(transferId); - nodeService.deleteNode(tempStoreNode); - log.debug("Deleted temporary store node."); - } - catch (Exception ex) - { - log.warn("Failed to delete temp store node for transfer id " + transferId + - "\nTemp store noderef = " + tempStoreNode); + log.debug("releasing lock:" + lock.lockToken); + lock.releaseLock(); + locks.remove(lock); } - File stagingFolder = null; - try - { - log.debug("delete staging folder " + transferId); - // Delete the staging folder. - stagingFolder = getStagingFolder(transferId); - deleteFile(stagingFolder); - log.debug("Staging folder deleted"); - } - catch(Exception ex) - { - log.warn("Failed to delete staging folder for transfer id " + transferId + - "\nStaging folder = " + stagingFolder.toString()); - } + removeTempFolders(transferId); + //Fire the OnEndInboundTransfer policy Set createdNodes = Collections.emptySet(); @@ -577,8 +629,42 @@ public class RepoTransferReceiverImpl implements TransferReceiver, } } + private void removeTempFolders(final String transferId) + { + NodeRef tempStoreNode = null; + try + { + log.debug("Deleting temporary store node..."); + tempStoreNode = getTempFolder(transferId); + nodeService.deleteNode(tempStoreNode); + log.debug("Deleted temporary store node."); + } + catch (Exception ex) + { + log.warn("Failed to delete temp store node for transfer id " + transferId + + "\nTemp store noderef = " + tempStoreNode); + } + + File stagingFolder = null; + try + { + log.debug("delete staging folder " + transferId); + // Delete the staging folder. + stagingFolder = getStagingFolder(transferId); + deleteFile(stagingFolder); + log.debug("Staging folder deleted"); + } + catch(Exception ex) + { + log.warn("Failed to delete staging folder for transfer id " + transferId + + "\nStaging folder = " + stagingFolder.toString()); + } + } + + public void cancel(String transferId) throws TransferException { + // no need to check the lock TransferProgress progress = getProgressMonitor().getProgress(transferId); getProgressMonitor().updateStatus(transferId, TransferProgress.Status.CANCELLED); if (progress.getStatus().equals(TransferProgress.Status.PRE_COMMIT)) @@ -589,6 +675,17 @@ public class RepoTransferReceiverImpl implements TransferReceiver, public void prepare(String transferId) throws TransferException { + // Check that this transfer still owns the lock + Lock lock = checkLock(transferId); + try + { + + } + finally + { + lock.enableLockTimeout(); + } + } /** @@ -609,58 +706,41 @@ public class RepoTransferReceiverImpl implements TransferReceiver, } file.delete(); } - - private NodeRef getLockNode() - { - final NodeRef lockFolder = getLockFolder(); - List assocs = nodeService.getChildAssocs(lockFolder, ContentModel.ASSOC_CONTAINS, - LOCK_QNAME); - NodeRef lockId = assocs.size() == 0 ? null : assocs.get(0).getChildRef(); - return lockId; - } - - private boolean testLockedTransfer(NodeRef lockId, String transferId) - { - if (lockId == null) - { - throw new IllegalArgumentException("lockId = null"); - } - if (transferId == null) - { - throw new IllegalArgumentException("transferId = null"); - } - String currentTransferId = (String) nodeService.getProperty(lockId, TransferModel.PROP_TRANSFER_ID); - // Check that the lock is held for the specified transfer (error if not) - return (transferId.equals(currentTransferId)); - } - + /* * (non-Javadoc) * * @see org.alfresco.service.cmr.transfer.TransferReceiver#nudgeLock(java.lang.String) */ - public void nudgeLock(final String transferId) throws TransferException + public Lock checkLock(final String transferId) throws TransferException { if (transferId == null) - throw new IllegalArgumentException("transferId = null"); - - transactionService.getRetryingTransactionHelper().doInTransaction( - new RetryingTransactionHelper.RetryingTransactionCallback() - { - public NodeRef execute() throws Throwable - { - // Find the lock node - NodeRef lockId = getLockNode(); - // Check that the specified transfer is the one that owns the lock - if (!testLockedTransfer(lockId, transferId)) - { - throw new TransferException(MSG_NOT_LOCK_OWNER); - } - // Just write the lock file name again (no change, but forces the modified time to be updated) - nodeService.setProperty(lockId, ContentModel.PROP_NAME, LOCK_FILE_NAME); - return null; - } - }, false, true); + { + throw new IllegalArgumentException("nudgeLock: transferId = null"); + } + + Lock lock = locks.get(transferId); + if(lock != null) + { + if(lock.isActive()) + { + lock.suspendLockTimeout(); + return lock; + } + else + { + // lock is no longer active + log.debug("lock not active"); + throw new TransferException(MSG_LOCK_TIMED_OUT, new Object[]{transferId}); + + } + } + else + { + log.debug("lock not found"); + throw new TransferException(MSG_LOCK_NOT_FOUND, new Object[]{transferId}); + // lock not found + } } /* @@ -670,28 +750,35 @@ public class RepoTransferReceiverImpl implements TransferReceiver, */ public void saveSnapshot(String transferId, InputStream openStream) throws TransferException { - // Check that this transfer owns the lock and give it a nudge to stop it expiring - nudgeLock(transferId); - - if (log.isDebugEnabled()) - { - log.debug("Saving snapshot for transferId =" + transferId); - } - File snapshotFile = new File(getStagingFolder(transferId), SNAPSHOT_FILE_NAME); + // Check that this transfer still owns the lock + Lock lock = checkLock(transferId); try { - if (snapshotFile.createNewFile()) - { - FileCopyUtils.copy(openStream, new FileOutputStream(snapshotFile)); - } if (log.isDebugEnabled()) { - log.debug("Saved snapshot for transferId =" + transferId); + log.debug("Saving snapshot for transferId =" + transferId); + } + + File snapshotFile = new File(getStagingFolder(transferId), SNAPSHOT_FILE_NAME); + try + { + if (snapshotFile.createNewFile()) + { + FileCopyUtils.copy(openStream, new BufferedOutputStream(new FileOutputStream(snapshotFile))); + } + if (log.isDebugEnabled()) + { + log.debug("Saved snapshot for transferId =" + transferId); + } + } + catch (Exception ex) + { + throw new TransferException(MSG_ERROR_WHILE_STAGING_SNAPSHOT, new Object[]{transferId}, ex); } } - catch (Exception ex) + finally { - throw new TransferException(MSG_ERROR_WHILE_STAGING_SNAPSHOT, new Object[]{transferId}, ex); + lock.enableLockTimeout(); } } @@ -704,10 +791,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver, public void saveContent(String transferId, String contentFileId, InputStream contentStream) throws TransferException { - nudgeLock(transferId); - File stagedFile = new File(getStagingFolder(transferId), contentFileId); + Lock lock = checkLock(transferId); try - { + { + File stagedFile = new File(getStagingFolder(transferId), contentFileId); if (stagedFile.createNewFile()) { FileCopyUtils.copy(contentStream, new BufferedOutputStream(new FileOutputStream(stagedFile))); @@ -717,20 +804,47 @@ public class RepoTransferReceiverImpl implements TransferReceiver, { throw new TransferException(MSG_ERROR_WHILE_STAGING_CONTENT, new Object[]{transferId, contentFileId}, ex); } + finally + { + lock.enableLockTimeout(); + } } public void commitAsync(String transferId) { - nudgeLock(transferId); - progressMonitor.updateStatus(transferId, Status.COMMIT_REQUESTED); - Action commitAction = actionService.createAction(TransferCommitActionExecuter.NAME); - commitAction.setParameterValue(TransferCommitActionExecuter.PARAM_TRANSFER_ID, transferId); - commitAction.setExecuteAsynchronously(true); - actionService.executeAction(commitAction, new NodeRef(transferId)); - if (log.isDebugEnabled()) + /** + * A side-effect of checking the lock here is that the lock timeout is suspended. + * + */ + Lock lock = checkLock(transferId); + try { - log.debug("Registered transfer commit for asynchronous execution: " + transferId); + progressMonitor.updateStatus(transferId, Status.COMMIT_REQUESTED); + Action commitAction = actionService.createAction(TransferCommitActionExecuter.NAME); + commitAction.setParameterValue(TransferCommitActionExecuter.PARAM_TRANSFER_ID, transferId); + commitAction.setExecuteAsynchronously(true); + actionService.executeAction(commitAction, new NodeRef(transferId)); + if (log.isDebugEnabled()) + { + log.debug("Registered transfer commit for asynchronous execution: " + transferId); + } } + catch (Exception error) + { + /** + * Error somewhere in the action service? + */ + //TODO consider whether the methods in this class should be retried/retryable.. + + // need to re-enable the lock timeout otherwise we will hold the lock forever... + lock.enableLockTimeout(); + + throw new TransferException(MSG_ERROR_WHILE_COMMITTING_TRANSFER, new Object[]{transferId}, error); + } + + /** + * Lock intentionally not re-enabled here + */ } public void commit(final String transferId) throws TransferException @@ -740,6 +854,11 @@ public class RepoTransferReceiverImpl implements TransferReceiver, log.debug("Committing transferId=" + transferId); } + /** + * A side-effect of checking the lock here is that it ensures that the lock timeout is suspended. + */ + checkLock(transferId); + /** * Turn off rules while transfer is being committed. */ @@ -748,7 +867,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver, try { - nudgeLock(transferId); + /* lock is going to be released */ checkLock(transferId); progressMonitor.updateStatus(transferId, Status.COMMITTING); RetryingTransactionHelper.RetryingTransactionCallback commitWork = new RetryingTransactionCallback() @@ -788,7 +907,6 @@ public class RepoTransferReceiverImpl implements TransferReceiver, // behaviourFilter.enableBehaviour(ContentModel.ASPECT_AUDITABLE); behaviourFilter.enableAllBehaviours(); } - nudgeLock(transferId); parser.reset(); } } @@ -1264,4 +1382,229 @@ public class RepoTransferReceiverImpl implements TransferReceiver, nodeService.setProperty(nodeRef, TransferModel.PROP_FROM_CONTENT, null); } } + + public void setJobLockService(JobLockService jobLockService) + { + this.jobLockService = jobLockService; + } + + public JobLockService getJobLockService() + { + return jobLockService; + } + + public void setLockRetryCount(int lockRetryCount) + { + this.lockRetryCount = lockRetryCount; + } + + public int getLockRetryCount() + { + return lockRetryCount; + } + + public void setLockRetryWait(long lockRetryWait) + { + this.lockRetryWait = lockRetryWait; + } + + public long getLockRetryWait() + { + return lockRetryWait; + } + + public void setLockTimeOut(long lockTimeOut) + { + this.lockTimeOut = lockTimeOut; + } + + public long getLockTimeOut() + { + return lockTimeOut; + } + + public void setLockRefreshTime(long lockRefreshTime) + { + this.lockRefreshTime = lockRefreshTime; + } + + public long getLockRefreshTime() + { + return lockRefreshTime; + } + + /** + * A Transfer Lock + */ + private class Lock implements JobLockService.JobLockRefreshCallback + { + /** + * The name of the lock - unique for each domain + */ + QName lockQName; + + /** + * The unique token for this lock instance. + */ + String lockToken; + + /** + * The transfer that this lock belongs to. + */ + String transferId; + + /** + * Is the lock active ? + */ + private boolean active = false; + + /** + * Is the server processing ? + */ + private boolean processing = false; + + /** + * When did we last check whether the lock is active + */ + Date lastActive = new Date(); + + public Lock(QName lockQName) + { + this.lockQName = lockQName; + } + + + /** + * Make the lock - called on main thread + * + * @throws LockAquisitionException + */ + public void makeLock() + { + if(log.isDebugEnabled()) + { + log.debug("makeLock" + lockQName); + } + + lockToken = getJobLockService().getLock(lockQName, getLockRefreshTime(), getLockRetryWait(), getLockRetryCount()); + + synchronized(this) + { + active = true; + } + + if (log.isDebugEnabled()) + { + log.debug("lock taken: name" + lockQName + " token:" +lockToken); + } + log.debug("register lock callback, target lock refresh time :" + getLockRefreshTime()); + getJobLockService().refreshLock(lockToken, lockQName, getLockRefreshTime(), this); + log.debug("refreshLock callback registered"); + } + + /** + * Check that the lock is still active + * + * Called on main transfer thread as transfer proceeds. + * @throws TransferException (Lock timeout) + */ + public void suspendLockTimeout() + { + log.debug("suspend lock called"); + if(active) + { + processing = true; + } + else + { + // lock is no longer active + log.debug("lock not active, throw timed out exception"); + throw new TransferException(MSG_LOCK_TIMED_OUT); + } + } + + public void enableLockTimeout() + { + Date now = new Date(); + + // Update lastActive to 1S boundary + if(now.getTime() > lastActive.getTime() + 1000) + { + lastActive = new Date(); + log.debug("start waiting : lastActive:" + lastActive); + } + + processing = false; + } + + /** + * Release the lock + * + * Called on main thread + */ + public void releaseLock() + { + if(log.isDebugEnabled()) + { + log.debug("transfer service about to releaseLock : " + lockQName); + } + + synchronized(this) + { + if(active) + { + getJobLockService().releaseLock(lockToken, lockQName); + } + active = false; + } + } + + /** + * Called by Job Lock Service to determine whether the lock is still active + */ + @Override + public boolean isActive() + { + Date now = new Date(); + + synchronized(this) + { + if(active) + { + if(!processing) + { + if(now.getTime() > lastActive.getTime() + getLockTimeOut()) + { + return false; + } + } + } + + if(log.isDebugEnabled()) + { + log.debug("transfer service callback isActive: " + active); + } + + return active; + } + } + + /** + * Called by Job Lock Service on release of the lock after time-out + */ + @Override + public void lockReleased() + { + synchronized(this) + { + if(active) + { + log.info("transfer service: lock has timed out, timeout :" + lockQName); + timeout(transferId); + } + + active = false; + } + } + } } diff --git a/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImplTest.java b/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImplTest.java index 3fd2834506..b929b8170a 100644 --- a/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImplTest.java +++ b/source/java/org/alfresco/repo/transfer/RepoTransferReceiverImplTest.java @@ -38,6 +38,8 @@ import org.alfresco.repo.policy.JavaBehaviour; import org.alfresco.repo.policy.PolicyComponent; import org.alfresco.repo.policy.Behaviour.NotificationFrequency; import org.alfresco.repo.security.authentication.AuthenticationComponent; +import org.alfresco.repo.transaction.RetryingTransactionHelper; +import org.alfresco.repo.transaction.RetryingTransactionHelper.RetryingTransactionCallback; import org.alfresco.repo.transfer.manifest.TransferManifestDeletedNode; import org.alfresco.repo.transfer.manifest.TransferManifestHeader; import org.alfresco.repo.transfer.manifest.TransferManifestNode; @@ -179,44 +181,174 @@ public class RepoTransferReceiverImplTest extends BaseAlfrescoSpringTest } + /** + * Tests start and end with regard to locking. + * @throws Exception + */ public void testStartAndEnd() throws Exception { log.info("testStartAndEnd"); - startNewTransaction(); + + RetryingTransactionHelper trx = transactionService.getRetryingTransactionHelper(); + + RetryingTransactionCallback cb = new RetryingTransactionCallback() + { + + @Override + public Void execute() throws Throwable + { + log.debug("about to call start"); + String transferId = receiver.start(); + File stagingFolder = null; + try + { + System.out.println("TransferId == " + transferId); + + stagingFolder = receiver.getStagingFolder(transferId); + assertTrue(receiver.getStagingFolder(transferId).exists()); + NodeRef tempFolder = receiver.getTempFolder(transferId); + assertNotNull("tempFolder is null", tempFolder); + + Thread.sleep(1000); + try + { + receiver.start(); + fail("Successfully started twice!"); + } + catch (TransferException ex) + { + // Expected + } + + Thread.sleep(300); + try + { + receiver.start(); + fail("Successfully started twice!"); + } + catch (TransferException ex) + { + // Expected + } + + try + { + receiver.end(new NodeRef(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, GUID.generate()).toString()); +// fail("Successfully ended with transfer id that doesn't own lock."); + } + catch (TransferException ex) + { + // Expected + } + } + finally + { + log.debug("about to call end"); + receiver.end(transferId); + + /** + * Check clean-up + */ + if(stagingFolder != null) + { + assertFalse(stagingFolder.exists()); + } + } + + return null; + } + }; + + long oldRefreshTime = receiver.getLockRefreshTime(); try { - String transferId = receiver.start(); - System.out.println("TransferId == " + transferId); - - File stagingFolder = receiver.getStagingFolder(transferId); - assertTrue(receiver.getStagingFolder(transferId).exists()); - - try + receiver.setLockRefreshTime(500); + + for (int i = 0; i < 5; i++) { - receiver.start(); - fail("Successfully started twice!"); + log.info("test iteration:" + i); + trx.doInTransaction(cb, false, true); } - catch (TransferException ex) - { - // Expected - } - try - { - receiver.end(new NodeRef(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, GUID.generate()).toString()); - fail("Successfully ended with transfer id that doesn't own lock."); - } - catch (TransferException ex) - { - // Expected - } - receiver.end(transferId); - assertFalse(stagingFolder.exists()); - - receiver.end(receiver.start()); } finally { - endTransaction(); + receiver.setLockRefreshTime(oldRefreshTime); + } + } + + /** + * Tests start and end with regard to locking. + * + * Going to cut down the timeout to a very short period, the lock should expire + * @throws Exception + */ + public void testLockTimeout() throws Exception + { + log.info("testStartAndEnd"); + + RetryingTransactionHelper trx = transactionService.getRetryingTransactionHelper(); + + /** + * Simulates a client starting a transfer and then "going away"; + */ + RetryingTransactionCallback startWithoutAnythingElse = new RetryingTransactionCallback() + { + @Override + public Void execute() throws Throwable + { + log.debug("about to call start"); + String transferId = receiver.start(); + return null; + } + }; + + RetryingTransactionCallback slowTransfer = new RetryingTransactionCallback() + { + @Override + public Void execute() throws Throwable + { + log.debug("about to call start"); + String transferId = receiver.start(); + Thread.sleep(1000); + try + { + receiver.saveSnapshot(transferId, null); + fail("did not timeout"); + } + catch (TransferException te) + { + logger.debug("expected to timeout", te); + // expect to go here with a timeout + } + return null; + } + }; + + + long lockRefreshTime = receiver.getLockRefreshTime(); + long lockTimeOut = receiver.getLockTimeOut(); + + try + { + receiver.setLockRefreshTime(500); + receiver.setLockTimeOut(200); + + /** + * This test simulates a client that starts a transfer and then "goes away". + * We kludge the timeouts to far shorter than normal to make the test run in a reasonable time. + */ + for (int i = 0; i < 3; i++) + { + log.info("test iteration:" + i); + trx.doInTransaction(startWithoutAnythingElse, false, true); + Thread.sleep(1000); + } + trx.doInTransaction(slowTransfer, false, true); + } + finally + { + receiver.setLockRefreshTime(lockRefreshTime); + receiver.setLockTimeOut(lockTimeOut); } } diff --git a/source/java/org/alfresco/repo/transfer/TransferServiceImplTest.java b/source/java/org/alfresco/repo/transfer/TransferServiceImplTest.java index 5c8fc1e09b..b1e79c4e1d 100644 --- a/source/java/org/alfresco/repo/transfer/TransferServiceImplTest.java +++ b/source/java/org/alfresco/repo/transfer/TransferServiceImplTest.java @@ -22,6 +22,8 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.io.BufferedReader; +import java.io.File; +import java.io.FileWriter; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; @@ -92,6 +94,7 @@ import org.alfresco.util.BaseAlfrescoSpringTest; import org.alfresco.util.GUID; import org.alfresco.util.Pair; import org.alfresco.util.PropertyMap; +import org.alfresco.util.TempFileProvider; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.util.ResourceUtils; @@ -1666,77 +1669,74 @@ public class TransferServiceImplTest extends BaseAlfrescoSpringTest */ public void testAsyncCallback() throws Exception { - int MAX_SLEEPS = 5; + final int MAX_SLEEPS = 5; + final RetryingTransactionHelper tran = transactionService.getRetryingTransactionHelper(); + /** - * Get guest home - */ - String guestHomeQuery = "/app:company_home/app:guest_home"; - ResultSet guestHomeResult = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, guestHomeQuery); - assertEquals("", 1, guestHomeResult.length()); - NodeRef guestHome = guestHomeResult.getNodeRef(0); + * Unit test kludge to transfer from guest home to company home + */ + final UnitTestTransferManifestNodeFactory testNodeFactory = unitTestKludgeToTransferGuestHomeToCompanyHome(); /** - * For unit test - * - replace the HTTP transport with the in-process transport - * - replace the node factory with one that will map node refs, paths etc. - */ - TransferTransmitter transmitter = new UnitTestInProcessTransmitterImpl(this.receiver, this.contentService, transactionService); - transferServiceImpl.setTransmitter(transmitter); - UnitTestTransferManifestNodeFactory testNodeFactory = new UnitTestTransferManifestNodeFactory(this.transferManifestNodeFactory); - transferServiceImpl.setTransferManifestNodeFactory(testNodeFactory); - List> pathMap = testNodeFactory.getPathMap(); - // Map company_home/guest_home to company_home so tranferred nodes and moved "up" one level. - pathMap.add(new Pair(PathHelper.stringToPath(GUEST_HOME_XPATH_QUERY), PathHelper.stringToPath(COMPANY_HOME_XPATH_QUERY))); - - DescriptorService mockedDescriptorService = getMockDescriptorService(REPO_ID_A); - transferServiceImpl.setDescriptorService(mockedDescriptorService); - - /** - * Now go ahead and create our first transfer target * This needs to be committed before we can call transfer asycnc. */ - String CONTENT_TITLE = "ContentTitle"; - String CONTENT_NAME_A = "Demo Node A"; - String CONTENT_NAME_B = "Demo Node B"; - Locale CONTENT_LOCALE = Locale.GERMAN; - String CONTENT_STRING = "Hello"; + final String CONTENT_TITLE = "ContentTitle"; + final String CONTENT_NAME_A = "Demo Node A"; + final String CONTENT_NAME_B = "Demo Node B"; + final Locale CONTENT_LOCALE = Locale.GERMAN; + final String CONTENT_STRING = "Hello"; - NodeRef nodeRefA = null; - NodeRef nodeRefB = null; - String targetName = "testAsyncCallback"; + final String targetName = "testAsyncCallback"; + class TestContext { - UserTransaction trx = transactionService.getNonPropagatingUserTransaction(); - trx.begin(); - try + TransferTarget transferMe; + NodeRef nodeRefA = null; + NodeRef nodeRefB = null; + }; + + RetryingTransactionCallback setupCB = new RetryingTransactionCallback() + { + @Override + public TestContext execute() throws Throwable { - nodeRefA = nodeService.getChildByName(guestHome, ContentModel.ASSOC_CONTAINS, CONTENT_NAME_A); + TestContext ctx = new TestContext(); - if(nodeRefA == null) + /** + * Get guest home + */ + String guestHomeQuery = "/app:company_home/app:guest_home"; + ResultSet guestHomeResult = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, guestHomeQuery); + assertEquals("", 1, guestHomeResult.length()); + final NodeRef guestHome = guestHomeResult.getNodeRef(0); + + ctx.nodeRefA = nodeService.getChildByName(guestHome, ContentModel.ASSOC_CONTAINS, CONTENT_NAME_A); + + if(ctx.nodeRefA == null) { /** * Create a test node that we will read and write */ ChildAssociationRef child = nodeService.createNode(guestHome, ContentModel.ASSOC_CONTAINS, QName.createQName(GUID.generate()), ContentModel.TYPE_CONTENT); - nodeRefA = child.getChildRef(); - nodeService.setProperty(nodeRefA, ContentModel.PROP_TITLE, CONTENT_TITLE); - nodeService.setProperty(nodeRefA, ContentModel.PROP_NAME, CONTENT_NAME_A); + ctx.nodeRefA = child.getChildRef(); + nodeService.setProperty(ctx.nodeRefA, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(ctx.nodeRefA, ContentModel.PROP_NAME, CONTENT_NAME_A); - ContentWriter writer = contentService.getWriter(nodeRefA, ContentModel.PROP_CONTENT, true); + ContentWriter writer = contentService.getWriter(ctx.nodeRefA, ContentModel.PROP_CONTENT, true); writer.setLocale(CONTENT_LOCALE); writer.putContent(CONTENT_STRING); } - nodeRefB = nodeService.getChildByName(guestHome, ContentModel.ASSOC_CONTAINS, CONTENT_NAME_B); + ctx.nodeRefB = nodeService.getChildByName(guestHome, ContentModel.ASSOC_CONTAINS, CONTENT_NAME_B); - if(nodeRefB == null) + if(ctx.nodeRefB == null) { ChildAssociationRef child = nodeService.createNode(guestHome, ContentModel.ASSOC_CONTAINS, QName.createQName(GUID.generate()), ContentModel.TYPE_CONTENT); - nodeRefB = child.getChildRef(); - nodeService.setProperty(nodeRefB, ContentModel.PROP_TITLE, CONTENT_TITLE); - nodeService.setProperty(nodeRefB, ContentModel.PROP_NAME, CONTENT_NAME_B); + ctx.nodeRefB = child.getChildRef(); + nodeService.setProperty(ctx.nodeRefB, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(ctx.nodeRefB, ContentModel.PROP_NAME, CONTENT_NAME_B); - ContentWriter writer = contentService.getWriter(nodeRefB, ContentModel.PROP_CONTENT, true); + ContentWriter writer = contentService.getWriter(ctx.nodeRefB, ContentModel.PROP_CONTENT, true); writer.setLocale(CONTENT_LOCALE); writer.putContent(CONTENT_STRING); } @@ -1751,33 +1751,28 @@ public class TransferServiceImplTest extends BaseAlfrescoSpringTest else { transferService.getTransferTarget(targetName); - } - } - finally - { - trx.commit(); - } - } + } + + return ctx; + } + }; - /** - * The transfer report is a plain report of the transfer - no async shenanigans to worry about - */ - ListtransferReport = new ArrayList(50); - - startNewTransaction(); - try - { - /** - * Call the transferAsync method. - */ + final TestContext testContext = tran.doInTransaction(setupCB); + + RetryingTransactionCallback> transferCB = new RetryingTransactionCallback>() { + + @Override + public List execute() throws Throwable { + ListtransferReport = new ArrayList(50); + TestTransferCallback callback = new TestTransferCallback(); Set callbacks = new HashSet(); callbacks.add(callback); TransferDefinition definition = new TransferDefinition(); Setnodes = new HashSet(); - nodes.add(nodeRefA); - nodes.add(nodeRefB); + nodes.add(testContext.nodeRefA); + nodes.add(testContext.nodeRefB); definition.setNodes(nodes); transferService.transferAsync(targetName, definition, callbacks); @@ -1850,32 +1845,32 @@ public class TransferServiceImplTest extends BaseAlfrescoSpringTest event = events.poll(); } } - } - - /** - * Now validate the transferReport - */ - assertTrue("transfer report is too small", transferReport.size() > 2); - assertTrue("transfer report does not start with START", transferReport.get(0).getTransferState().equals(TransferEvent.TransferState.START)); - boolean success = false; - for(TransferEvent event : transferReport) - { - if(event.getTransferState() == TransferEvent.TransferState.SUCCESS) - { - success = true; - } + return transferReport; } - //assertTrue("transfer report does not contain SUCCESS", success)); - } - finally - { -// UserTransaction trx = transactionService.getNonPropagatingUserTransaction(); -// trx.begin(); - transferService.deleteTransferTarget(targetName); -// trx.commit(); - endTransaction(); - } + }; + + /** + * The transfer report is a plain report of the transfer - no async shenanigans to worry about + */ + final ListtransferReport = tran.doInTransaction(transferCB); + + /** + * Now validate the transferReport + */ + assertTrue("transfer report is too small", transferReport.size() > 2); + assertTrue("transfer report does not start with START", transferReport.get(0).getTransferState().equals(TransferEvent.TransferState.START)); + + boolean success = false; + for(TransferEvent event : transferReport) + { + if(event.getTransferState() == TransferEvent.TransferState.SUCCESS) + { + success = true; + } + } + assertTrue("transfer report does not contain SUCCESS", success); + } // test async callback @@ -2523,93 +2518,170 @@ public class TransferServiceImplTest extends BaseAlfrescoSpringTest } } + private UnitTestTransferManifestNodeFactory unitTestKludgeToTransferGuestHomeToCompanyHome() + { + /** + * For unit test + * - replace the HTTP transport with the in-process transport + * - replace the node factory with one that will map node refs, paths etc. + */ + TransferTransmitter transmitter = new UnitTestInProcessTransmitterImpl(this.receiver, this.contentService, transactionService); + transferServiceImpl.setTransmitter(transmitter); + UnitTestTransferManifestNodeFactory testNodeFactory = new UnitTestTransferManifestNodeFactory(this.transferManifestNodeFactory); + transferServiceImpl.setTransferManifestNodeFactory(testNodeFactory); + List> pathMap = testNodeFactory.getPathMap(); + // Map company_home/guest_home to company_home so tranferred nodes and moved "up" one level. + pathMap.add(new Pair(PathHelper.stringToPath(GUEST_HOME_XPATH_QUERY), PathHelper.stringToPath(COMPANY_HOME_XPATH_QUERY))); + + DescriptorService mockedDescriptorService = getMockDescriptorService(REPO_ID_A); + transferServiceImpl.setDescriptorService(mockedDescriptorService); + + return testNodeFactory; + + } -// /** -// * Test the transfer method with big content - commented out since it takes a long time to run. -// */ -// public void testTransferOneNodeWithBigContent() throws Exception -// { -// String CONTENT_TITLE = "ContentTitle"; -// String CONTENT_NAME = "Demo Node 6"; -// Locale CONTENT_LOCALE = Locale.GERMAN; -// String CONTENT_STRING = "Hello"; -// -// String targetName = "testTransferOneNodeWithBigContent"; -// -// String guestHomeQuery = "/app:company_home/app:guest_home"; -// ResultSet result = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, guestHomeQuery); -// -// assertEquals("", 1, result.length()); -// NodeRef guestHome = result.getNodeRef(0); -// ChildAssociationRef childAssoc = result.getChildAssocRef(0); -// System.out.println("Guest home:" + guestHome); -// assertNotNull(guestHome); -// -// /** -// * Now go ahead and create our first transfer target -// */ -// TransferTarget transferMe = createTransferTarget(targetName); -// -// /** -// * Create a test node that we will read and write -// */ -// ChildAssociationRef child = nodeService.createNode(guestHome, ContentModel.ASSOC_CONTAINS, QName.createQName("testNode6"), ContentModel.TYPE_CONTENT); -// -// -// File tempFile = TempFileProvider.createTempFile("test", ".dat"); -// FileWriter fw = new FileWriter(tempFile); -// for(int i = 0; i < 100000000; i++) -// { -// fw.write("hello world this is my text, I wonder how much text I can transfer?" + i); -// } -// System.out.println("Temp File Size is:" + tempFile.length()); -// fw.close(); -// -// NodeRef contentNodeRef = child.getChildRef(); -// ContentWriter writer = contentService.getWriter(contentNodeRef, ContentModel.PROP_CONTENT, true); -// writer.setLocale(CONTENT_LOCALE); -// //File file = new File("c:/temp/images/BigCheese1.bmp"); -// writer.setMimetype("application/data"); -// //writer.putContent(file); -// writer.putContent(tempFile); -// -// tempFile.delete(); -// -// nodeService.setProperty(contentNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); -// nodeService.setProperty(contentNodeRef, ContentModel.PROP_NAME, CONTENT_NAME); -// -// try -// { -// /** -// * Transfer the node created above -// */ -// { -// TransferDefinition definition = new TransferDefinition(); -// Setnodes = new HashSet(); -// nodes.add(contentNodeRef); -// definition.setNodes(nodes); -// transferService.transfer(targetName, definition, null); -// } -// -// /** -// * Negative test transfer nothing -// */ -// try -// { -// TransferDefinition definition = new TransferDefinition(); -// transferService.transfer(targetName, definition, null); -// fail("exception not thrown"); -// } -// catch(TransferException te) -// { -// // expect to go here -// } -// } -// finally -// { -// transferService.deleteTransferTarget(targetName); -// } -// } + /** + * Test the transfer method with regard to big content. + * + * This test takes a long time to run and is by default not run in the overnight build. + * + * Turn it on by turning debug logging on for this class or by changing the "runTest" value; + */ + public void testTransferOneNodeWithBigContent() throws Exception + { + /** + * This test takes a long time to run - so switch it on and off here. + */ + boolean runTest = false; + if(runTest || logger.isDebugEnabled()) + { + final String CONTENT_TITLE = "ContentTitle"; + final String CONTENT_NAME = "BigContent"; + final Locale CONTENT_LOCALE = Locale.UK; + + logger.debug("testTransferOneNodeWithBigContent starting"); + + final RetryingTransactionHelper tran = transactionService.getRetryingTransactionHelper(); + + /** + * Unit test kludge to transfer from guest home to company home + */ + final UnitTestTransferManifestNodeFactory testNodeFactory = unitTestKludgeToTransferGuestHomeToCompanyHome(); + + final String targetName = "testTransferOneNodeWithBigContent"; + + class TestContext + { + TransferTarget transferMe; + NodeRef contentNodeRef; + NodeRef destNodeRef; + }; + + RetryingTransactionCallback setupCB = new RetryingTransactionCallback() + { + @Override + public TestContext execute() throws Throwable + { + TestContext ctx = new TestContext(); + + String guestHomeQuery = "/app:company_home/app:guest_home"; + ResultSet result = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, guestHomeQuery); + + assertEquals("", 1, result.length()); + NodeRef guestHome = result.getNodeRef(0); + + System.out.println("Guest home:" + guestHome); + assertNotNull(guestHome); + + ctx.contentNodeRef = nodeService.getChildByName(guestHome, ContentModel.ASSOC_CONTAINS, CONTENT_NAME); + if(ctx.contentNodeRef == null) + { + /** + * Create a test node that we will read and write + */ + ChildAssociationRef child = nodeService.createNode(guestHome, ContentModel.ASSOC_CONTAINS, QName.createQName(CONTENT_NAME), ContentModel.TYPE_CONTENT); + + File tempFile = TempFileProvider.createTempFile("test", ".dat"); + FileWriter fw = new FileWriter(tempFile); + for(int i = 0; i < 100000000; i++) + { + fw.write("hello world this is my text, I wonder how much text I can transfer?" + i); + } + System.out.println("Temp File Size is:" + tempFile.length()); + fw.close(); + + ctx.contentNodeRef = child.getChildRef(); + ContentWriter writer = contentService.getWriter(ctx.contentNodeRef, ContentModel.PROP_CONTENT, true); + writer.setLocale(CONTENT_LOCALE); + writer.setMimetype("application/data"); + writer.putContent(tempFile); + + tempFile.delete(); + + nodeService.setProperty(ctx.contentNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(ctx.contentNodeRef, ContentModel.PROP_NAME, CONTENT_NAME); + } + if(!transferService.targetExists(targetName)) + { + createTransferTarget(targetName); + } + + return ctx; + } + }; + + final TestContext testContext = tran.doInTransaction(setupCB); + + RetryingTransactionCallback transferCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + TransferDefinition definition = new TransferDefinition(); + Setnodes = new HashSet(); + nodes.add(testContext.contentNodeRef); + definition.setNodes(nodes); + transferService.transfer(targetName, definition); + + return null; + } + }; + + RetryingTransactionCallback finishCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + NodeRef oldDestNodeRef = testNodeFactory.getMappedNodeRef(testContext.contentNodeRef); + + ContentReader source = contentService.getReader(testContext.contentNodeRef, ContentModel.PROP_CONTENT); + ContentReader destination = contentService.getReader(oldDestNodeRef, ContentModel.PROP_CONTENT); + + assertNotNull("source is null", source); + assertNotNull("destination is null", destination); + assertEquals("size different", source.getSize(), destination.getSize()); + + /** + * Now get rid of the transferred node so that the test can run again. + */ + nodeService.deleteNode(oldDestNodeRef); + + return null; + } + }; + + /** + * This is the test + */ + tran.doInTransaction(transferCB); + tran.doInTransaction(finishCB); + + } + else + { + System.out.println("test supressed"); + } + } // test big content /** * Test the transfer method behaviour with respect to sync folders - sending a complete set @@ -7383,6 +7455,183 @@ public class TransferServiceImplTest extends BaseAlfrescoSpringTest } } // test repeat update content + /** + * Test the transfer method with regard to replacing a node. ALF-5109 + * + * Step 1: Create a new parent node and child node + * transfer + * + * Step 2: Delete the parent node + * transfer + * + * Step 3: Create new parent child node with same names and assocs. + * transfer + * + * This is a unit test so it does some shenanigans to send to the same instance of alfresco. + */ + public void testReplaceNode() throws Exception + { + final RetryingTransactionHelper tran = transactionService.getRetryingTransactionHelper(); + + final String CONTENT_TITLE = "ContentTitle"; + final String CONTENT_TITLE_UPDATED = "ContentTitleUpdated"; + final Locale CONTENT_LOCALE = Locale.GERMAN; + final String CONTENT_STRING = "Hello World"; + final String CONTENT_UPDATE_STRING = "Foo Bar"; + + /** + * For unit test + * - replace the HTTP transport with the in-process transport + * - replace the node factory with one that will map node refs, paths etc. + * + * Fake Repository Id + */ + TransferTransmitter transmitter = new UnitTestInProcessTransmitterImpl(receiver, contentService, transactionService); + transferServiceImpl.setTransmitter(transmitter); + UnitTestTransferManifestNodeFactory testNodeFactory = new UnitTestTransferManifestNodeFactory(this.transferManifestNodeFactory); + transferServiceImpl.setTransferManifestNodeFactory(testNodeFactory); + List> pathMap = testNodeFactory.getPathMap(); + // Map company_home/guest_home to company_home so tranferred nodes and moved "up" one level. + pathMap.add(new Pair(PathHelper.stringToPath(GUEST_HOME_XPATH_QUERY), PathHelper.stringToPath(COMPANY_HOME_XPATH_QUERY))); + + DescriptorService mockedDescriptorService = getMockDescriptorService(REPO_ID_A); + transferServiceImpl.setDescriptorService(mockedDescriptorService); + + /** + * Get guest home + */ + String guestHomeQuery = "/app:company_home/app:guest_home"; + ResultSet guestHomeResult = searchService.query(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, SearchService.LANGUAGE_XPATH, guestHomeQuery); + assertEquals("", 1, guestHomeResult.length()); + final NodeRef guestHome = guestHomeResult.getNodeRef(0); + + final String targetName = "testRepeatUpdateOfContent"; + + class TestContext + { + TransferTarget transferMe; + NodeRef parentNodeRef; + NodeRef middleNodeRef; + NodeRef childNodeRef; + QName parentName; + QName middleName; + QName childName; + + }; + + RetryingTransactionCallback setupCB = new RetryingTransactionCallback() + { + @Override + public TestContext execute() throws Throwable + { + TestContext testContext = new TestContext(); + + /** + * Create a test node that we will read and write + */ + String name = GUID.generate(); + + testContext.parentName = QName.createQName(name); + testContext.childName = QName.createQName("Ermintrude"); + testContext.middleName = QName.createQName("Matilda"); + + ChildAssociationRef child = nodeService.createNode(guestHome, ContentModel.ASSOC_CONTAINS, testContext.parentName, ContentModel.TYPE_FOLDER); + testContext.parentNodeRef = child.getChildRef(); + logger.debug("parentNodeRef created:" + testContext.parentNodeRef ); + nodeService.setProperty(testContext.parentNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(testContext.parentNodeRef, ContentModel.PROP_NAME, testContext.parentName.getLocalName()); + + ChildAssociationRef child2 = nodeService.createNode(testContext.parentNodeRef, ContentModel.ASSOC_CONTAINS, testContext.childName, ContentModel.TYPE_FOLDER); + testContext.middleNodeRef = child2.getChildRef(); + logger.debug("middleNodeRef created:" + testContext.middleNodeRef ); + nodeService.setProperty(testContext.middleNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(testContext.middleNodeRef, ContentModel.PROP_NAME, testContext.childName.getLocalName()); + + ChildAssociationRef child3 = nodeService.createNode(testContext.middleNodeRef, ContentModel.ASSOC_CONTAINS, testContext.childName, ContentModel.TYPE_CONTENT); + testContext.childNodeRef = child3.getChildRef(); + logger.debug("childNodeRef created:" + testContext.childNodeRef ); + nodeService.setProperty(testContext.childNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(testContext.childNodeRef, ContentModel.PROP_NAME, testContext.childName.getLocalName()); + + /** + * Make sure the transfer target exists and is enabled. + */ + if(!transferService.targetExists(targetName)) + { + testContext.transferMe = createTransferTarget(targetName); + } + else + { + testContext.transferMe = transferService.getTransferTarget(targetName); + } + transferService.enableTransferTarget(targetName, true); + return testContext; + } + }; + + final TestContext testContext = tran.doInTransaction(setupCB); + + RetryingTransactionCallback transferCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + TransferDefinition definition = new TransferDefinition(); + Collection nodes = new ArrayList(); + nodes.add(testContext.childNodeRef); + nodes.add(testContext.parentNodeRef); + nodes.add(testContext.middleNodeRef); + definition.setSync(true); + definition.setNodes(nodes); + transferService.transfer(targetName, definition); + return null; + } + }; + + RetryingTransactionCallback checkTransferCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + return null; + } + }; + + RetryingTransactionCallback replaceNodesCB = new RetryingTransactionCallback() { + + @Override + public Void execute() throws Throwable + { + // Delete the old nodes + + nodeService.deleteNode(testContext.middleNodeRef); + logger.debug("deleted node"); + + ChildAssociationRef child2 = nodeService.createNode(testContext.parentNodeRef, ContentModel.ASSOC_CONTAINS, testContext.childName, ContentModel.TYPE_FOLDER); + testContext.middleNodeRef = child2.getChildRef(); + logger.debug("middleNodeRef created:" + testContext.middleNodeRef ); + nodeService.setProperty(testContext.middleNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(testContext.middleNodeRef, ContentModel.PROP_NAME, testContext.childName.getLocalName()); + + ChildAssociationRef child3 = nodeService.createNode(testContext.middleNodeRef, ContentModel.ASSOC_CONTAINS, testContext.childName, ContentModel.TYPE_CONTENT); + testContext.childNodeRef = child3.getChildRef(); + logger.debug("childNodeRef created:" + testContext.childNodeRef ); + nodeService.setProperty(testContext.childNodeRef, ContentModel.PROP_TITLE, CONTENT_TITLE); + nodeService.setProperty(testContext.childNodeRef, ContentModel.PROP_NAME, testContext.childName.getLocalName()); + + return null; + } + }; + + // This is the test + + tran.doInTransaction(transferCB); + tran.doInTransaction(replaceNodesCB); + tran.doInTransaction(transferCB); + tran.doInTransaction(checkTransferCB); + + } // test replace node + private void createUser(String userName, String password) { diff --git a/source/java/org/alfresco/service/cmr/transfer/TransferReceiver.java b/source/java/org/alfresco/service/cmr/transfer/TransferReceiver.java index 9b6e970b2c..5e2dc0ad6e 100644 --- a/source/java/org/alfresco/service/cmr/transfer/TransferReceiver.java +++ b/source/java/org/alfresco/service/cmr/transfer/TransferReceiver.java @@ -1,19 +1,19 @@ /* - * Copyright (C) 2009-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 + * Copyright (C) 2009-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 . */ @@ -60,13 +60,6 @@ public interface TransferReceiver */ void end(String transferId) throws TransferException; - /** - * Nudge the transfer lock (to prevent it expiring) if the supplied transferId matches that referenced by the lock. - * @param transferId - * @throws TransferException if the lock doesn't exist or doesn't correspond to the supplied transferId. - */ - void nudgeLock(String transferId) throws TransferException; - /** * Store the specified snapshot file into the transfer staging area. * The specified transfer must currently be the holder of the transfer lock, otherwise an exception is thrown.