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

Merge BRANCHES/DEV/BRIAN/FSTR to HEAD:

Browse Source 
28906: File Transfer Receiver SPRINT1
 29365: Initial check in
 29378: Configuration properties can now be located in "filetransferreceiver.properties" separated from spring configuration files.
 29417: Delete, first check in
 29450: Now multiple distinct contents can be send over in one transfer.
 29516: 
 29517: 
 29531: Manage cases like /F1/F2/F3 ... becoming /F3/F2/F1
 29550: FTR:
    - Tidied project classpath
    - Removed incorrect reference to OpenOffice RuntimeException in JobLockServiceImpl
    - Removed references to StringOutputStream



git-svn-id: https://svn.alfresco.com/repos/alfresco-enterprise/alfresco/HEAD/root@29643 c4b6b30b-aa2e-2d43-bbcb-ca4b014f7261
This commit is contained in:
Brian Remmington
2011-08-09 21:02:03 +00:00
parent 62cfc60ce9
commit 153e2f8d33
10 changed files with 412 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
config/alfresco/model/transferModel.xml
View File

@@ -96,6 +96,9 @@
</properties>
</type>
<type name="trx:transferLock">
<title>Transfer Lock</title>
<description>Node type used to represent the transfer lock node
@@ -160,7 +163,7 @@
<description>Transfer Report</description>
<parent>cm:content</parent>
</type>
<type name="trx:transferReportDest">
<title>Transfer Report From Destination</title>
<description>Transfer Report (Destination)</description>
@@ -182,6 +185,24 @@
</properties>
</aspect>
<aspect name="trx:fileTransferTarget">
<title>File Transfer Target</title>
<description>Aspect used to model a receiving root for file transfer</description>
<associations>
<association name="trx:rootFileTransfer">
<source>
<mandatory>false</mandatory>
<many>true</many>
</source>
<target>
<class>cm:folder</class>
<mandatory>false</mandatory>
<many>true</many>
</target>
</association>
</associations>
</aspect>
<aspect name="trx:transferRelated">
<title>Nodes with this aspect are related to a particular transfer.
</title>
@@ -194,7 +215,7 @@
</property>
</properties>
</aspect>
<aspect name="trx:transferred">
<title>Transferred Node</title>
<description>Nodes with this aspect have been transferred from one repository to another</description>
@@ -220,7 +241,7 @@
</property>
</properties>
</aspect>
<aspect name="trx:alien">
<title>Alien Node</title>
<description>Nodes with this aspect are either alien nodes or have been invaded by other alien nodes</description>
@@ -235,7 +256,7 @@
</property>
</properties>
</aspect>
</aspects>
</model>

1
config/alfresco/transfer-service-context.xml
View File

@@ -54,6 +54,7 @@
<bean id="transferTransmitter" class="org.alfresco.repo.transfer.HttpClientTransmitterImpl"
init-method="init">
<property name="contentService" ref="ContentService" />
<property name="nodeService" ref="nodeService" />
</bean>
<bean id="transferVersionChecker" class="org.alfresco.repo.transfer.TransferVersionCheckerImpl">

2
source/java/org/alfresco/repo/lock/JobLockServiceImpl.java
View File

@@ -37,8 +37,6 @@ import org.alfresco.util.VmShutdownListener;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.sun.star.uno.RuntimeException;
/**
* {@inheritDoc JobLockService}
*

261
source/java/org/alfresco/repo/transfer/HttpClientTransmitterImpl.java
View File

@@ -31,12 +31,18 @@ import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import org.alfresco.service.cmr.repository.AssociationRef;
import org.alfresco.service.cmr.repository.ContentData;
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.transfer.TransferException;
import org.alfresco.service.cmr.transfer.TransferProgress;
import org.alfresco.service.cmr.transfer.TransferTarget;
@@ -68,34 +74,35 @@ import org.json.JSONObject;
/**
* HTTP implementation of TransferTransmitter.
*
*
* Sends data via HTTP to the server.
*
* @author brian
*
* @author brian
*/
public class HttpClientTransmitterImpl implements TransferTransmitter
{
private static final Log log = LogFactory.getLog(HttpClientTransmitterImpl.class);
private static final String MSG_UNSUPPORTED_PROTOCOL = "transfer_service.comms.unsupported_protocol";
private static final String MSG_UNSUCCESSFUL_RESPONSE = "transfer_service.comms.unsuccessful_response";
private static final String MSG_HTTP_REQUEST_FAILED = "transfer_service.comms.http_request_failed";
private static final int DEFAULT_HTTP_PORT = 80;
private static final int DEFAULT_HTTPS_PORT = 443;
private static final String HTTP_SCHEME_NAME = "http"; // lowercase is important
private static final String HTTPS_SCHEME_NAME = "https"; // lowercase is important
private HttpClient httpClient = null;
private Protocol httpProtocol = new Protocol(HTTP_SCHEME_NAME, new DefaultProtocolSocketFactory(), DEFAULT_HTTP_PORT);
private Protocol httpsProtocol = new Protocol(HTTPS_SCHEME_NAME, (ProtocolSocketFactory) new SSLProtocolSocketFactory(), DEFAULT_HTTPS_PORT);
private Map<String,Protocol> protocolMap = null;
private HttpMethodFactory httpMethodFactory = null;
private JsonSerializer<Throwable, JSONObject> jsonErrorSerializer;
private ContentService contentService;
private NodeService nodeService;
public HttpClientTransmitterImpl()
{
protocolMap = new TreeMap<String,Protocol>();
@@ -107,7 +114,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
httpMethodFactory = new StandardHttpMethodFactoryImpl();
jsonErrorSerializer = new ExceptionJsonSerializer();
}
public void init()
{
PropertyCheck.mandatory(this, "contentService", contentService);
@@ -115,7 +122,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
/**
* By default this class uses the standard SSLProtocolSocketFactory, but this method allows this to be overridden.
* Useful if, for example, one wishes to permit support of self-signed certificates on the target.
* Useful if, for example, one wishes to permit support of self-signed certificates on the target.
* @param socketFactory
*/
public void setHttpsSocketFactory(ProtocolSocketFactory socketFactory)
@@ -124,7 +131,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
}
/**
* By default, this class uses a plain HttpClient instance with the only non-default
* By default, this class uses a plain HttpClient instance with the only non-default
* option being the multi-threaded connection manager.
* Use this method to replace this with your own HttpClient instance configured how you wish
* @param httpClient
@@ -144,24 +151,24 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
verifyRequest.setPath(target.getEndpointPath() + "/test");
try
{
int response = httpClient.executeMethod(hostConfig, verifyRequest, httpState);
checkResponseStatus("verifyTarget", response, verifyRequest);
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"verifyTraget", target.toString(), e.toString()}, e);
}
}
}
finally
{
verifyRequest.releaseConnection();
@@ -176,14 +183,14 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
if (response != 200)
{
Throwable error = null;
try
try
{
log.error("Received \"unsuccessful\" response code from target server: " + response);
String errorPayload = method.getResponseBodyAsString();
JSONObject errorObj = new JSONObject(errorPayload);
error = rehydrateError(errorObj);
}
catch (Exception ex)
}
catch (Exception ex)
{
throw new TransferException(MSG_UNSUCCESSFUL_RESPONSE, new Object[] {methodName, response});
}
@@ -206,8 +213,8 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
protected HttpState getHttpState(TransferTarget target)
{
HttpState httpState = new HttpState();
httpState.setCredentials(new AuthScope(target.getEndpointHost(), target.getEndpointPort(),
AuthScope.ANY_REALM),
httpState.setCredentials(new AuthScope(target.getEndpointHost(), target.getEndpointPort(),
AuthScope.ANY_REALM),
new UsernamePasswordCredentials(target.getUsername(), new String(target.getPassword())));
return httpState;
}
@@ -223,7 +230,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
throw new TransferException(MSG_UNSUPPORTED_PROTOCOL, new Object[] {target.getEndpointProtocol()});
}
Protocol protocol = protocolMap.get(requiredProtocol.toLowerCase().trim());
if (protocol == null) {
log.error("Unsupported protocol: " + target.getEndpointProtocol());
@@ -242,32 +249,45 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
beginRequest.setPath(target.getEndpointPath() + "/begin");
try
{
beginRequest.setRequestBody( new NameValuePair[] {
NameValuePair[] nameValuePair = new NameValuePair[] {
new NameValuePair(TransferCommons.PARAM_FROM_REPOSITORYID, fromRepositoryId),
new NameValuePair(TransferCommons.PARAM_ALLOW_TRANSFER_TO_SELF, "false"),
new NameValuePair(TransferCommons.PARAM_VERSION_EDITION, fromVersion.getEdition()),
new NameValuePair(TransferCommons.PARAM_VERSION_MAJOR, fromVersion.getVersionMajor()),
new NameValuePair(TransferCommons.PARAM_VERSION_MINOR, fromVersion.getVersionMinor()),
new NameValuePair(TransferCommons.PARAM_VERSION_REVISION, fromVersion.getVersionRevision()),
new NameValuePair(TransferCommons.PARAM_VERSION_EDITION, fromVersion.getEdition())
});
new NameValuePair(TransferCommons.PARAM_VERSION_REVISION, fromVersion.getVersionRevision())
};
//add the parameter defining the root of the transfer on the file system if exist
NodeRef transferRootNode = this.getFileTransferRootNodeRef(target.getNodeRef());
if (transferRootNode != null)
{
//add the parameter
ArrayList<NameValuePair> nameValuePairArrayList= new ArrayList<NameValuePair>(nameValuePair.length + 1);
Collections.addAll(nameValuePairArrayList,nameValuePair);
nameValuePairArrayList.add(new NameValuePair(TransferCommons.PARAM_ROOT_FILE_TRANSFER, transferRootNode.toString()));
nameValuePair = nameValuePairArrayList.toArray(new NameValuePair[0]);
}
beginRequest.setRequestBody(nameValuePair);
int responseStatus = httpClient.executeMethod(hostConfig, beginRequest, httpState);
checkResponseStatus("begin", responseStatus, beginRequest);
//If we get here then we've received a 200 response
//We're expecting the transfer id encoded in a JSON object...
JSONObject response = new JSONObject(beginRequest.getResponseBodyAsString());
Transfer transfer = new Transfer();
transfer.setTransferTarget(target);
String transferId = response.getString(TransferCommons.PARAM_TRANSFER_ID);
transfer.setTransferId(transferId);
if(response.has(TransferCommons.PARAM_VERSION_MAJOR))
{
String versionMajor = response.getString(TransferCommons.PARAM_VERSION_MAJOR);
@@ -282,26 +302,26 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
TransferVersion version = new TransferVersionImpl("0", "0", "0", "Unknown");
transfer.setToVersion(version);
}
if(log.isDebugEnabled())
{
log.debug("begin transfer transferId:" + transferId +", target:" + target);
}
return transfer;
}
}
catch (RuntimeException e)
{
log.debug("unexpected exception", e);
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[] {"begin", target.toString(), e.toString()}, e);
}
}
}
finally
{
log.debug("releasing connection");
@@ -314,38 +334,38 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
TransferTarget target = transfer.getTransferTarget();
PostMethod postSnapshotRequest = getPostMethod();
MultipartRequestEntity requestEntity;
if(log.isDebugEnabled())
{
log.debug("does manifest exist? " + manifest.exists());
log.debug("sendManifest file : " + manifest.getAbsoluteFile());
}
try
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
try
{
postSnapshotRequest.setPath(target.getEndpointPath() + "/post-snapshot");
//Put the transferId on the query string
postSnapshotRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
//TODO encapsulate the name of the manifest part
//And add the manifest file as a "part"
Part file = new FilePart(TransferCommons.PART_NAME_MANIFEST, manifest);
requestEntity = new MultipartRequestEntity(new Part[] {file}, postSnapshotRequest.getParams());
postSnapshotRequest.setRequestEntity(requestEntity);
int responseStatus = httpClient.executeMethod(hostConfig, postSnapshotRequest, httpState);
checkResponseStatus("sendManifest", responseStatus, postSnapshotRequest);
InputStream is = postSnapshotRequest.getResponseBodyAsStream();
final ReadableByteChannel inputChannel = Channels.newChannel(is);
final WritableByteChannel outputChannel = Channels.newChannel(result);
try
@@ -358,20 +378,20 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
inputChannel.close();
outputChannel.close();
}
return;
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"sendManifest", target.toString(), e.toString()}, e);
}
}
}
finally
{
postSnapshotRequest.releaseConnection();
@@ -386,30 +406,30 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
abortRequest.setPath(target.getEndpointPath() + "/abort");
//Put the transferId on the query string
abortRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
try
{
int responseStatus = httpClient.executeMethod(hostConfig, abortRequest, httpState);
checkResponseStatus("abort", responseStatus, abortRequest);
//If we get here then we've received a 200 response
//We're expecting the transfer id encoded in a JSON object...
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"abort", target.toString(), e.toString()}, e);
}
}
}
finally
{
abortRequest.releaseConnection();
@@ -424,7 +444,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
commitRequest.setPath(target.getEndpointPath() + "/commit");
//Put the transferId on the query string
commitRequest.setQueryString(
@@ -435,18 +455,18 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
checkResponseStatus("commit", responseStatus, commitRequest);
//If we get here then we've received a 200 response
//We're expecting the transfer id encoded in a JSON object...
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.error(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"commit", target.toString(), e.toString()}, e);
}
}
}
finally
{
commitRequest.releaseConnection();
@@ -461,7 +481,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
prepareRequest.setPath(target.getEndpointPath() + "/prepare");
//Put the transferId on the query string
prepareRequest.setQueryString(
@@ -472,18 +492,18 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
checkResponseStatus("prepare", responseStatus, prepareRequest);
//If we get here then we've received a 200 response
//We're expecting the transfer id encoded in a JSON object...
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"prepare", target.toString(), e.toString()}, e);
}
}
}
finally
{
prepareRequest.releaseConnection();
@@ -491,7 +511,7 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
}
/**
*
*
*/
public void sendContent(Transfer transfer, Set<ContentData> data) throws TransferException
{
@@ -514,54 +534,54 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
//Put the transferId on the query string
postContentRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
//Put the transferId on the query string
postContentRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
Part[] parts = new Part[data.size()];
int index = 0;
for(ContentData content : data)
{
{
String contentUrl = content.getContentUrl();
String fileName = TransferCommons.URLToPartName(contentUrl);
log.debug("content partName: " + fileName);
parts[index++] = new ContentDataPart(getContentService(), fileName, content);
parts[index++] = new ContentDataPart(getContentService(), fileName, content);
}
MultipartRequestEntity requestEntity = new MultipartRequestEntity(parts, postContentRequest.getParams());
postContentRequest.setRequestEntity(requestEntity);
int responseStatus = httpClient.executeMethod(hostConfig, postContentRequest, httpState);
checkResponseStatus("sendContent", responseStatus, postContentRequest);
if(log.isDebugEnabled())
{
log.debug("sent content");
}
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"sendContent", target.toString(), e.toString()}, e);
}
}
}
finally
{
postContentRequest.releaseConnection();
}
} // end of sendContent
/**
*
*
*/
public TransferProgress getStatus(Transfer transfer) throws TransferException
{
@@ -571,12 +591,12 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
statusRequest.setPath(target.getEndpointPath() + "/status");
//Put the transferId on the query string
statusRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
try
{
int responseStatus = httpClient.executeMethod(hostConfig, statusRequest, httpState);
@@ -584,45 +604,45 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
//If we get here then we've received a 200 response
String statusPayload = statusRequest.getResponseBodyAsString();
JSONObject statusObj = new JSONObject(statusPayload);
//We're expecting the transfer progress encoded in a JSON object...
//We're expecting the transfer progress encoded in a JSON object...
int currentPosition = statusObj.getInt("currentPosition");
int endPosition = statusObj.getInt("endPosition");
String statusStr= statusObj.getString("status");
TransferProgress p = new TransferProgress();
if(statusObj.has("error"))
{
JSONObject errorJSON = statusObj.getJSONObject("error");
Throwable throwable = rehydrateError(errorJSON);
p.setError(throwable);
}
p.setStatus(TransferProgress.Status.valueOf(statusStr));
p.setCurrentPosition(currentPosition);
p.setEndPosition(endPosition);
return p;
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"status", target.toString(), e.toString()}, e);
}
}
}
finally
{
statusRequest.releaseConnection();
}
}
/**
*
*
*/
public void getTransferReport(Transfer transfer, OutputStream result)
{
@@ -632,20 +652,20 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
HostConfiguration hostConfig = getHostConfig(target);
HttpState httpState = getHttpState(target);
try
{
getReportRequest.setPath(target.getEndpointPath() + "/report");
//Put the transferId on the query string
getReportRequest.setQueryString(
new NameValuePair[] {new NameValuePair("transferId", transfer.getTransferId())});
int responseStatus = httpClient.executeMethod(hostConfig, getReportRequest, httpState);
checkResponseStatus("getReport", responseStatus, getReportRequest);
InputStream is = getReportRequest.getResponseBodyAsStream();
// Now copy the response input stream to result.
final ReadableByteChannel inputChannel = Channels.newChannel(is);
final WritableByteChannel outputChannel = Channels.newChannel(result);
@@ -660,46 +680,46 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
inputChannel.close();
outputChannel.close();
}
return;
}
}
catch (RuntimeException e)
{
throw e;
}
}
catch (Exception e)
{
String error = "Failed to execute HTTP request to target";
log.debug(error, e);
throw new TransferException(MSG_HTTP_REQUEST_FAILED, new Object[]{"getTransferReport", target.toString(), e.toString()}, e);
}
}
}
finally
{
getReportRequest.releaseConnection();
}
}
}
private static void channelCopy(final ReadableByteChannel src, final WritableByteChannel dest) throws IOException
private static void channelCopy(final ReadableByteChannel src, final WritableByteChannel dest) throws IOException
{
final ByteBuffer buffer = ByteBuffer.allocateDirect(2 * 1024);
while (src.read(buffer) != -1)
while (src.read(buffer) != -1)
{
// prepare the buffer to be drained
buffer.flip();
// write to the channel, may block
dest.write(buffer);
// If partial transfer, shift remainder down
// If buffer is empty, same as doing clear()
buffer.compact();
}
// EOF will leave buffer in fill state
buffer.flip();
// make sure the buffer is fully drained.
while (buffer.hasRemaining())
while (buffer.hasRemaining())
{
dest.write(buffer);
}
@@ -709,20 +729,20 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
{
return httpMethodFactory.createPostMethod();
}
/**
*
*
* @param errorJSON A JSON object expected to hold the name of the error class ("errorType"),
* the error message ("errorMessage"), and, optionally, the Alfresco message id ("alfrescoErrorId")
* and Alfresco message parameters ("alfrescoErrorParams").
* @return The rehydrated error object, or null if errorJSON is null.
* @throws JSONException if an error occurs while parsing the supplied JSON object
*/
private Throwable rehydrateError(JSONObject errorJSON)
private Throwable rehydrateError(JSONObject errorJSON)
{
return jsonErrorSerializer.deserialize(errorJSON);
}
public void setContentService(ContentService contentService)
{
this.contentService = contentService;
@@ -741,5 +761,26 @@ public class HttpClientTransmitterImpl implements TransferTransmitter
public void setJsonErrorSerializer(JsonSerializer<Throwable, JSONObject> jsonErrorSerializer)
{
this.jsonErrorSerializer = jsonErrorSerializer;
}
}
public void setNodeService(NodeService nodeService)
{
this.nodeService = nodeService;
}
private NodeRef getFileTransferRootNodeRef(NodeRef transferNodeRef)
{
//testing if transferring to file system
if(!nodeService.hasAspect(transferNodeRef, TransferModel.ASPECT_FILE_TRANSFER_TARGET))
return null;
//get association
List<AssociationRef> assocs = nodeService.getTargetAssocs(transferNodeRef, TransferModel.ASSOC_ROOT_FILE_TRANSFER);
if(assocs.size() == 0 || assocs.size() > 1)
return null;
return assocs.get(0).getTargetRef();
}
}

315
source/java/org/alfresco/repo/transfer/RepoTransferReceiverImpl.java
View File

@@ -106,7 +106,7 @@ import org.springframework.util.FileCopyUtils;
*
* @author brian
*/
public class RepoTransferReceiverImpl implements TransferReceiver,
public class RepoTransferReceiverImpl implements TransferReceiver,
NodeServicePolicies.OnCreateChildAssociationPolicy,
NodeServicePolicies.BeforeDeleteNodePolicy,
NodeServicePolicies.OnRestoreNodePolicy,
@@ -115,9 +115,9 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
/**
* This embedded class is used to push requests for asynchronous commits onto a different thread
*
*
* @author Brian
*
*
*/
public class AsyncCommitCommand implements Runnable
{
@@ -171,7 +171,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
private static final String MSG_LOCK_NOT_FOUND = "transfer_service.receiver.lock_not_found";
private static final String MSG_TRANSFER_TO_SELF = "transfer_service.receiver.error.transfer_to_self";
private static final String MSG_INCOMPATIBLE_VERSIONS = "transfer_service.incompatible_versions";
private static final String SNAPSHOT_FILE_NAME = "snapshot.xml";
private NodeService nodeService;
@@ -192,49 +192,49 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
private AlienProcessor alienProcessor;
private JobLockService jobLockService;
private TransferVersionChecker transferVersionChecker;
/**
* Where the temporary files are stored. Tenant Domain Name, NodeRef
* Where the temporary files are stored. Tenant Domain Name, NodeRef
*/
private Map<String,NodeRef> transferTempFolderMap = new ConcurrentHashMap<String, NodeRef>();
/**
* Where the destination side transfer report is generated. Tenant Domain Name, NodeRef
*/
private Map<String,NodeRef> inboundTransferRecordsFolderMap = new ConcurrentHashMap<String, NodeRef>();
private ClassPolicyDelegate<BeforeStartInboundTransferPolicy> beforeStartInboundTransferDelegate;
private ClassPolicyDelegate<OnStartInboundTransferPolicy> onStartInboundTransferDelegate;
private ClassPolicyDelegate<OnEndInboundTransferPolicy> onEndInboundTransferDelegate;
/**
* Locks for the transfers in progress
* <p>
* TransferId, Lock
*/
private Map<String, Lock> locks = new ConcurrentHashMap<String, Lock>();
private Map<String, Lock> locks = new ConcurrentHashMap<String, Lock>();
/**
* 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);
@@ -262,15 +262,15 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
*/
this.getPolicyComponent().bindAssociationBehaviour(
NodeServicePolicies.OnCreateChildAssociationPolicy.QNAME,
TransferModel.ASPECT_TRANSFERRED,
TransferModel.ASPECT_TRANSFERRED,
new JavaBehaviour(this, "onCreateChildAssociation", NotificationFrequency.EVERY_EVENT));
/**
* For every update of a transferred node
*/
* For every update of a transferred node
*/
this.getPolicyComponent().bindClassBehaviour(
ContentServicePolicies.OnContentUpdatePolicy.QNAME,
TransferModel.ASPECT_TRANSFERRED,
TransferModel.ASPECT_TRANSFERRED,
new JavaBehaviour(this, "onContentUpdate", NotificationFrequency.EVERY_EVENT));
/**
@@ -278,53 +278,53 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
*/
this.getPolicyComponent().bindClassBehaviour(
CopyServicePolicies.OnCopyNodePolicy.QNAME,
TransferModel.ASPECT_TRANSFERRED,
TransferModel.ASPECT_TRANSFERRED,
new JavaBehaviour(this, "onCopyTransferred", NotificationFrequency.EVERY_EVENT));
/**
* For every new child of a node with the trx:alien aspect run this.onCreateChildAssociation
*/
this.getPolicyComponent().bindAssociationBehaviour(
NodeServicePolicies.OnCreateChildAssociationPolicy.QNAME,
TransferModel.ASPECT_ALIEN,
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "onCreateChildAssociation", NotificationFrequency.EVERY_EVENT));
/**
* For every node with the trx:alien aspect run this.beforeDeleteNode
*/
this.getPolicyComponent().bindClassBehaviour(
NodeServicePolicies.BeforeDeleteNodePolicy.QNAME,
NodeServicePolicies.BeforeDeleteNodePolicy.QNAME,
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "beforeDeleteNode", NotificationFrequency.EVERY_EVENT));
new JavaBehaviour(this, "beforeDeleteNode", NotificationFrequency.EVERY_EVENT));
/**
* For every restore of a node with the trx:alien aspect
*/
this.getPolicyComponent().bindClassBehaviour(
NodeServicePolicies.OnRestoreNodePolicy.QNAME,
TransferModel.ASPECT_ALIEN,
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "onRestoreNode", NotificationFrequency.EVERY_EVENT));
/**
* For every move of a node with the trx:alien aspect.
*/
this.getPolicyComponent().bindClassBehaviour(
NodeServicePolicies.OnMoveNodePolicy.QNAME,
TransferModel.ASPECT_ALIEN,
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "onMoveNode", NotificationFrequency.EVERY_EVENT));
/**
* For every copy of an alien node remove the alien aspect
*/
this.getPolicyComponent().bindClassBehaviour(
CopyServicePolicies.OnCopyNodePolicy.QNAME,
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "onCopyAlien", NotificationFrequency.EVERY_EVENT));
TransferModel.ASPECT_ALIEN,
new JavaBehaviour(this, "onCopyAlien", NotificationFrequency.EVERY_EVENT));
}
/*
* (non-Javadoc)
*
*
* @see
* org.alfresco.repo.web.scripts.transfer.TransferReceiver#getStagingFolder(org.alfresco.service.cmr.repository.
* NodeRef)
@@ -355,7 +355,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
String tenantDomain = tenantService.getUserDomain(AuthenticationUtil.getRunAsUser());
NodeRef transferTempFolder = transferTempFolderMap.get(tenantDomain);
// Have we already resolved the node that is the temp folder?
// If not then do so.
if (transferTempFolder == null)
@@ -401,28 +401,28 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
/*
* (non-Javadoc)
*
*
* @see org.alfresco.repo.web.scripts.transfer.TransferReceiver#start()
*/
public String start(String fromRepositoryId, boolean transferToSelf, TransferVersion fromVersion)
{
log.debug("Start transfer");
/**
* Check that transfer is allowed to this repository
*/
checkTransfer(fromRepositoryId, transferToSelf);
/**
* Check that the versions are compatible
*/
TransferVersion toVersion = getVersion();
if(!getTransferVersionChecker().checkTransferVersions(fromVersion, toVersion))
{
throw new TransferException(MSG_INCOMPATIBLE_VERSIONS, new Object[] {"None", fromVersion, toVersion});
}
/**
* First get the transfer lock for this domain
*/
@@ -430,15 +430,15 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
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
{
TransferServicePolicies.BeforeStartInboundTransferPolicy beforeStartPolicy =
TransferServicePolicies.BeforeStartInboundTransferPolicy beforeStartPolicy =
beforeStartInboundTransferDelegate.get(TransferModel.TYPE_TRANSFER_RECORD);
beforeStartPolicy.beforeStartInboundTransfer();
lock.makeLock();
/**
* Transfer Lock held if we get this far
*/
@@ -450,7 +450,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
* Now create a transfer record and use its NodeRef as the transfer id
*/
RetryingTransactionHelper txHelper = transactionService.getRetryingTransactionHelper();
transferId = txHelper.doInTransaction(
new RetryingTransactionHelper.RetryingTransactionCallback<String>()
{
@@ -461,10 +461,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
getTempFolder(transferId);
getStagingFolder(transferId);
TransferServicePolicies.OnStartInboundTransferPolicy onStartPolicy =
TransferServicePolicies.OnStartInboundTransferPolicy onStartPolicy =
onStartInboundTransferDelegate.get(TransferModel.TYPE_TRANSFER_RECORD);
onStartPolicy.onStartInboundTransfer(transferId);
return transferId;
}
}, false, true);
@@ -476,7 +476,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
lock.releaseLock();
throw new TransferException(MSG_ERROR_WHILE_STARTING, e);
}
/**
* Here if we have begun a transfer and have a valid transfer id
*/
@@ -485,7 +485,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.info("transfer started:" + transferId);
lock.enableLockTimeout();
return transferId;
}
catch (LockAcquisitionException lae)
{
@@ -525,7 +525,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
SimpleDateFormat format = new SimpleDateFormat("yyyyMMddHHmmssSSSZ");
String timeNow = format.format(new Date());
String name = timeNow + ".xml";
QName recordName = QName.createQName(NamespaceService.APP_MODEL_1_0_URI, name);
Map<QName, Serializable> props = new HashMap<QName, Serializable>();
@@ -541,33 +541,33 @@ 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
* There is no transaction or authentication context in this method since it is called via a
* timer thread.
*/
*/
final RetryingTransactionCallback<Void> timeoutCB = new RetryingTransactionCallback<Void>() {
@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
* 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);
@@ -578,28 +578,28 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
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"
// 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<String>()
{
public String doWork() throws Exception
{
transactionService.getRetryingTransactionHelper().doInTransaction(timeoutCB, false, true);
return null;
return null;
}
}, AuthenticationUtil.getSystemUserName());
}
/*
* (non-Javadoc)
*
*
* @see org.alfresco.repo.web.scripts.transfer.TransferReceiver#end(org.alfresco.service.cmr.repository.NodeRef)
*/
public void end(final String transferId)
@@ -622,9 +622,9 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
lock.releaseLock();
locks.remove(lock);
}
removeTempFolders(transferId);
//Fire the OnEndInboundTransfer policy
Set<NodeRef> createdNodes = Collections.emptySet();
@@ -637,7 +637,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
updatedNodes = new HashSet<NodeRef>(changesRecord.getUpdatedNodes());
deletedNodes = new HashSet<NodeRef>(changesRecord.getDeletedNodes());
}
TransferServicePolicies.OnEndInboundTransferPolicy onEndPolicy =
TransferServicePolicies.OnEndInboundTransferPolicy onEndPolicy =
onEndInboundTransferDelegate.get(TransferModel.TYPE_TRANSFER_RECORD);
onEndPolicy.onEndInboundTransfer(transferId, createdNodes, updatedNodes, deletedNodes);
}
@@ -663,10 +663,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
catch (Exception ex)
{
log.warn("Failed to delete temp store node for transfer id " + transferId +
log.warn("Failed to delete temp store node for transfer id " + transferId +
"\nTemp store noderef = " + tempStoreNode);
}
File stagingFolder = null;
try
{
@@ -678,7 +678,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
catch(Exception ex)
{
log.warn("Failed to delete staging folder for transfer id " + transferId +
log.warn("Failed to delete staging folder for transfer id " + transferId +
"\nStaging folder = " + stagingFolder.toString());
}
}
@@ -701,7 +701,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
Lock lock = checkLock(transferId);
try
{
}
finally
{
@@ -728,10 +728,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
file.delete();
}
/*
* (non-Javadoc)
*
*
* @see org.alfresco.service.cmr.transfer.TransferReceiver#nudgeLock(java.lang.String)
*/
public Lock checkLock(final String transferId) throws TransferException
@@ -740,7 +740,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
throw new IllegalArgumentException("nudgeLock: transferId = null");
}
Lock lock = locks.get(transferId);
if(lock != null)
{
@@ -754,7 +754,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
// lock is no longer active
log.debug("lock not active");
throw new TransferException(MSG_LOCK_TIMED_OUT, new Object[]{transferId});
}
}
else
@@ -767,7 +767,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
/*
* (non-Javadoc)
*
*
* @see org.alfresco.service.cmr.transfer.TransferReceiver#saveSnapshot(java.io.InputStream)
*/
public void saveSnapshot(String transferId, InputStream openStream) throws TransferException
@@ -780,7 +780,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
log.debug("Saving snapshot for transferId =" + transferId);
}
File snapshotFile = new File(getStagingFolder(transferId), SNAPSHOT_FILE_NAME);
try
{
@@ -806,7 +806,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
/*
* (non-Javadoc)
*
*
* @see org.alfresco.service.cmr.transfer.TransferReceiver#saveContent(java.lang.String, java.lang.String,
* java.io.InputStream)
*/
@@ -815,7 +815,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
Lock lock = checkLock(transferId);
try
{
{
File stagedFile = new File(getStagingFolder(transferId), contentFileId);
if (stagedFile.createNewFile())
{
@@ -836,7 +836,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
/**
* A side-effect of checking the lock here is that the lock timeout is suspended.
*
*
*/
Lock lock = checkLock(transferId);
try
@@ -857,13 +857,13 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
* 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
*/
@@ -875,18 +875,18 @@ 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.
*/
boolean rulesEnabled = ruleService.isEnabled();
ruleService.disableRules();
try
{
/* lock is going to be released */ checkLock(transferId);
@@ -919,7 +919,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
//behaviourFilter.disableBehaviour(ContentModel.ASPECT_AUDITABLE);
behaviourFilter.disableAllBehaviours();
try
{
parser.parse(snapshotFile, reader);
@@ -984,7 +984,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
*/
ruleService.enableRules();
}
/**
* Clean up at the end of the transfer
*/
@@ -1116,7 +1116,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
this.actionService = actionService;
}
/**
* Set the ruleService
* @param ruleService
@@ -1142,7 +1142,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
public void generateRequsite(String transferId, OutputStream out) throws TransferException
{
log.debug("Generate Requsite for transfer:" + transferId);
try
try
{
File snapshotFile = getSnapshotFile(transferId);
@@ -1150,12 +1150,12 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
log.debug("snapshot does exist");
SAXParserFactory saxParserFactory = SAXParserFactory.newInstance();
SAXParser parser = saxParserFactory.newSAXParser();
SAXParser parser = saxParserFactory.newSAXParser();
OutputStreamWriter dest = new OutputStreamWriter(out, "UTF-8");
XMLTransferRequsiteWriter writer = new XMLTransferRequsiteWriter(dest);
TransferManifestProcessor processor = manifestProcessorFactory.getRequsiteProcessor(
RepoTransferReceiverImpl.this,
RepoTransferReceiverImpl.this,
transferId,
writer);
@@ -1165,15 +1165,15 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
* Now run the parser
*/
parser.parse(snapshotFile, reader);
/**
* And flush the destination in case any content remains in the writer.
*/
dest.flush();
}
log.debug("Generate Requsite done transfer:" + transferId);
}
catch (Exception ex)
{
@@ -1187,7 +1187,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
}
}
public InputStream getTransferReport(String transferId)
{
return progressMonitor.getLogInputStream(transferId);
@@ -1205,21 +1205,21 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
/**
* When a new node is created as a child of a Transferred or Alien node then
* the new node needs to be marked as an alien.
* the new node needs to be marked as an alien.
* <p>
* Then the tree needs to be walked upwards to mark all parent
* Then the tree needs to be walked upwards to mark all parent
* transferred nodes as alien.
*/
public void onCreateChildAssociation(ChildAssociationRef childAssocRef,
boolean isNewNode)
{
log.debug("on create child association to transferred node");
final String localRepositoryId = descriptorService.getCurrentRepositoryDescriptor().getId();
alienProcessor.onCreateChild(childAssocRef, localRepositoryId, isNewNode);
alienProcessor.onCreateChild(childAssocRef, localRepositoryId, isNewNode);
}
/**
* When an alien node is deleted the it may be the last alien invader
* <p>
@@ -1230,9 +1230,9 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.debug("on delete node - need to check for transferred node");
alienProcessor.beforeDeleteAlien(deletedNodeRef, null);
}
/**
* When a transferred node is restored it may be a new invader or it may no
* When a transferred node is restored it may be a new invader or it may no
* longer be an invader.
* <p>
* Walk the tree checking the invasion status!
@@ -1243,22 +1243,22 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.debug("restoredAssocRef:" + childAssocRef);
alienProcessor.afterMoveAlien(childAssocRef);
}
/**
* When an alien node is moved it may un-invade its old location and invade a new
* When an alien node is moved it may un-invade its old location and invade a new
* location. The node may also cease to be alien.
*/
public void onMoveNode(ChildAssociationRef oldChildAssocRef,
ChildAssociationRef newChildAssocRef)
{
log.debug("onMoveNode");
log.debug("onMoveNode");
log.debug("oldChildAssocRef:" + oldChildAssocRef);
log.debug("newChildAssocRef:" + newChildAssocRef);
NodeRef oldParentRef = oldChildAssocRef.getParentRef();
NodeRef newParentRef = newChildAssocRef.getParentRef();
if(newParentRef.equals(oldParentRef))
{
log.debug("old parent and new parent are the same - this is a rename, do nothing");
@@ -1273,35 +1273,35 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
alienProcessor.afterMoveAlien(newChildAssocRef);
}
}
/**
* When a transferred node is copied, don't copy the transferred aspect.
*/
public CopyBehaviourCallback onCopyTransferred(QName classRef,
CopyDetails copyDetails)
{
return TransferredAspectCopyBehaviourCallback.INSTANCE;
return TransferredAspectCopyBehaviourCallback.INSTANCE;
}
/**
* When an alien node is copied, don't copy the alien aspect.
*/
public CopyBehaviourCallback onCopyAlien(QName classRef,
CopyDetails copyDetails)
{
return AlienAspectCopyBehaviourCallback.INSTANCE;
return AlienAspectCopyBehaviourCallback.INSTANCE;
}
/**
* Extends the default copy behaviour to prevent copying of transferred aspect and properties.
*
*
* @author Mark Rogers
* @since 3.4
*/
private static class TransferredAspectCopyBehaviourCallback extends DefaultCopyBehaviourCallback
{
private static final CopyBehaviourCallback INSTANCE = new TransferredAspectCopyBehaviourCallback();
/**
* @return Returns an empty map
*/
@@ -1311,10 +1311,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
return Collections.emptyMap();
}
/**
* Don't copy the transferred aspect.
*
*
* @return Returns <tt>true</tt> always
*/
@Override
@@ -1330,17 +1330,17 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
}
}
/**
* Extends the default copy behaviour to prevent copying of alien aspect and properties.
*
*
* @author Mark Rogers
* @since 3.4
*/
private static class AlienAspectCopyBehaviourCallback extends DefaultCopyBehaviourCallback
{
private static final CopyBehaviourCallback INSTANCE = new AlienAspectCopyBehaviourCallback();
/**
* @return Returns an empty map
*/
@@ -1350,10 +1350,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
return Collections.emptyMap();
}
/**
* Don't copy the transferred aspect.
*
*
* @return Returns <tt>true</tt> always
*/
@Override
@@ -1370,7 +1370,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
}
public void setDescriptorService(DescriptorService descriptorService)
{
this.descriptorService = descriptorService;
@@ -1380,7 +1380,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
return descriptorService;
}
public void setAlienProcessor(AlienProcessor alienProcessor)
{
this.alienProcessor = alienProcessor;
@@ -1404,7 +1404,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
nodeService.setProperty(nodeRef, TransferModel.PROP_FROM_CONTENT, null);
}
}
public void setJobLockService(JobLockService jobLockService)
{
this.jobLockService = jobLockService;
@@ -1464,41 +1464,41 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
* 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()
@@ -1507,14 +1507,14 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
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);
@@ -1523,10 +1523,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
getJobLockService().refreshLock(lockToken, lockQName, getLockRefreshTime(), this);
log.debug("refreshLock callback registered");
}
/**
* Check that the lock is still active
*
* Check that the lock is still active
*
* Called on main transfer thread as transfer proceeds.
* @throws TransferException (Lock timeout)
*/
@@ -1535,7 +1535,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.debug("suspend lock called");
if(active)
{
processing = true;
processing = true;
}
else
{
@@ -1544,24 +1544,24 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
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()
@@ -1570,7 +1570,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
{
log.debug("transfer service about to releaseLock : " + lockQName);
}
synchronized(this)
{
if(active)
@@ -1580,7 +1580,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
active = false;
}
}
/**
* Called by Job Lock Service to determine whether the lock is still active
*/
@@ -1588,7 +1588,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
public boolean isActive()
{
Date now = new Date();
synchronized(this)
{
if(active)
@@ -1601,7 +1601,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
}
}
if(log.isDebugEnabled())
{
log.debug("transfer service callback isActive: " + active);
@@ -1624,7 +1624,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.info("transfer service: lock has timed out, timeout :" + lockQName);
timeout(transferId);
}
active = false;
}
}
@@ -1642,7 +1642,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
log.debug("checkTransfer fromRepository:" + fromRepository + ", transferToSelf:" + transferToSelf );
}
final String localRepositoryId = descriptorService.getCurrentRepositoryDescriptor().getId();
if(!transferToSelf)
{
if(fromRepository != null)
@@ -1658,7 +1658,7 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
}
}
}
public void setTransferVersionChecker(TransferVersionChecker transferVersionChecker)
{
this.transferVersionChecker = transferVersionChecker;
@@ -1675,5 +1675,10 @@ public class RepoTransferReceiverImpl implements TransferReceiver,
Descriptor d = descriptorService.getServerDescriptor();
// needs to be serverDescriptor to pick up versionEdition
return new TransferVersionImpl(d);
}
}
public void setFileTransferRootNodeFileFileSystem(String rootFileSystem)
{
//just ignore, no relevant for transferring on file system
}
}

36
source/java/org/alfresco/repo/transfer/TransferCommons.java
View File

@@ -19,10 +19,10 @@
package org.alfresco.repo.transfer;
/**
* A bucket for little odds and ends for the transfer service.
*
* A bucket for little odds and ends for the transfer service.
*
* If this becomes a big class then refactor it away.
*
*
* @author Mark Rogers
*/
public class TransferCommons
@@ -31,45 +31,51 @@ public class TransferCommons
* The Mime Part Name of the manifest file
*/
public final static String PART_NAME_MANIFEST = "manifest";
/**
* The Query String for the begin method.
*/
public final static String PARAM_FROM_REPOSITORYID = "fromRepositoryId";
/**
* The Query String for the begin method.
*/
public final static String PARAM_ALLOW_TRANSFER_TO_SELF = "allowTransferToSelf";
/**
* TransferId
* TransferId
*/
public final static String PARAM_TRANSFER_ID = "transferId";
/**
* Major version
* Major version
*/
public final static String PARAM_VERSION_MAJOR = "versionMajor";
/**
* Minor version
*/
*/
public final static String PARAM_VERSION_MINOR = "versionMinor";
/**
* Revision version
* Revision version
*/
public final static String PARAM_VERSION_REVISION = "versionRevision";
/**
* Edition
*/
public final static String PARAM_VERSION_EDITION = "versionEdition";
/**
* File Root File Transfer
*/
public final static String PARAM_ROOT_FILE_TRANSFER = "rootFileTransfer";
/**
* Mapping between contentUrl and part name.
*
*
* @param URL
* @return the part name
*/

19
source/java/org/alfresco/repo/transfer/TransferModel.java
View File

@@ -22,7 +22,7 @@ import org.alfresco.service.namespace.QName;
/**
* Transfer Model Constants
*
*
* @author Mark Rogers
* @author Brian Remmington
*/
@@ -32,7 +32,7 @@ public interface TransferModel
static final QName ASPECT_ENABLEABLE = QName.createQName(TRANSFER_MODEL_1_0_URI, "enableable");
// static final QName ASSOC_IMAP_ATTACHMENTS_FOLDER = QName.createQName(IMAP_MODEL_1_0_URI, "attachmentsFolder");
/**
* Aspect : transferred
*/
@@ -47,11 +47,17 @@ public interface TransferModel
static final QName ASPECT_ALIEN = QName.createQName(TRANSFER_MODEL_1_0_URI, "alien");
static final QName PROP_INVADED_BY = QName.createQName(TRANSFER_MODEL_1_0_URI, "invadedBy");
/**
* Aspect : fileTransferTarget
*/
static final QName ASPECT_FILE_TRANSFER_TARGET = QName.createQName(TRANSFER_MODEL_1_0_URI, "fileTransferTarget");
static final QName ASSOC_ROOT_FILE_TRANSFER = QName.createQName(TRANSFER_MODEL_1_0_URI, "rootFileTransfer");
/*
* Type : Transfer Group
*/
static final QName TYPE_TRANSFER_GROUP = QName.createQName(TRANSFER_MODEL_1_0_URI, "transferGroup");
/*
* Type : Transfer Target
*/
@@ -62,7 +68,7 @@ public interface TransferModel
static final QName PROP_ENDPOINT_PATH = QName.createQName(TRANSFER_MODEL_1_0_URI, "endpointpath");
static final QName PROP_USERNAME = QName.createQName(TRANSFER_MODEL_1_0_URI, "username");
static final QName PROP_PASSWORD = QName.createQName(TRANSFER_MODEL_1_0_URI, "password");
static final QName PROP_ENABLED = QName.createQName(TRANSFER_MODEL_1_0_URI, "enabled");
/*
@@ -79,7 +85,7 @@ public interface TransferModel
static final QName PROP_PROGRESS_ENDPOINT = QName.createQName(TRANSFER_MODEL_1_0_URI, "progressEndpoint");
static final QName PROP_TRANSFER_STATUS = QName.createQName(TRANSFER_MODEL_1_0_URI, "transferStatus");
static final QName PROP_TRANSFER_ERROR = QName.createQName(TRANSFER_MODEL_1_0_URI, "transferError");
/*
* Type : Transfer report
*/
@@ -91,5 +97,6 @@ public interface TransferModel
*/
static final QName TYPE_TEMP_TRANSFER_STORE = QName.createQName(TRANSFER_MODEL_1_0_URI, "tempTransferStore");
static final QName ASSOC_TRANSFER_ORPHAN = QName.createQName(TRANSFER_MODEL_1_0_URI, "orphan");
}

20
source/java/org/alfresco/repo/transfer/TransferVersionImpl.java
View File

@@ -27,9 +27,9 @@ public class TransferVersionImpl implements TransferVersion
private String versionMinor;
private String versionRevision;
private String edition;
/**
*
*
* @param versionMajor
* @param versionMinor
* @param versionRevision
@@ -42,7 +42,7 @@ public class TransferVersionImpl implements TransferVersion
this.versionRevision = versionRevision;
this.edition = edition;
}
/**
* Construct a transferVersion from a system descriptor
* @param d the system descriptor
@@ -54,7 +54,7 @@ public class TransferVersionImpl implements TransferVersion
this.versionRevision = d.getVersionRevision();
this.edition = d.getEdition();
}
@Override
public String getVersionMajor()
{
@@ -78,7 +78,7 @@ public class TransferVersionImpl implements TransferVersion
{
return edition;
}
public String toString()
{
StringBuilder version = new StringBuilder();
@@ -89,10 +89,10 @@ public class TransferVersionImpl implements TransferVersion
version.append(getVersionMinor());
version.append(".");
version.append(getVersionRevision());
return version.toString();
}
public int hashCode()
{
if(edition != null && versionMinor != null)
@@ -104,18 +104,18 @@ public class TransferVersionImpl implements TransferVersion
return 1;
}
}
public boolean equals(Object other)
{
if(other == null)
{
return false;
}
if (other instanceof TransferVersion)
{
TransferVersion v = (TransferVersion)other;
if(!edition.equalsIgnoreCase(v.getEdition()))
{
return false;

40
source/java/org/alfresco/service/cmr/transfer/TransferReceiver.java
View File

@@ -33,19 +33,19 @@ import org.alfresco.service.cmr.repository.NodeRef;
public interface TransferReceiver
{
/**
*
*
* @param transferId
* @return
*/
File getStagingFolder(String transferId);
/**
*
*
* @param transferId
* @return
*/
NodeRef getTempFolder(String transferId);
/**
* Asks the receiver to setup a new transfer.
* @param fromRepositoryId the repositoryId of the sending system
@@ -66,14 +66,14 @@ public interface TransferReceiver
/**
* 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.
* This operation does not close the supplied stream, so the caller must do it as appropriate. The caller
* This operation does not close the supplied stream, so the caller must do it as appropriate. The caller
* should assume that the supplied stream has been fully read when this operation returns.
* @param transferId The identifier of the transfer with which this snapshot is associated
* @param snapshotStream The open stream that holds the snapshot file.
* @throws TransferException If an error occurs while saving the snapshot file.
*/
void saveSnapshot(String transferId, InputStream snapshotStream) throws TransferException;
/**
* Save a content item
* @param transferId
@@ -82,28 +82,28 @@ public interface TransferReceiver
* @throws TransferException
*/
void saveContent(String transferId, String contentId, InputStream contentStream) throws TransferException;
/**
* Write the requsite (the bits required to support the Manifest) to the output stream.
* @param requsiteStream an open stream to receive the requisite
* @throws TransferException
*/
void generateRequsite(String transferId, OutputStream requsiteStream) throws TransferException;
/**
* Prepare
* Prepare
* @param transferId
* @throws TransferException
*/
void prepare(String transferId) throws TransferException;
/**
* Abort
* @param transferId
* @throws TransferException
*/
void cancel(String transferId) throws TransferException;
/**
* Commit asynchronously
* @param transferId
@@ -117,29 +117,35 @@ public interface TransferReceiver
* @throws TransferException
*/
void commit(String transferId) throws TransferException;
/**
*
*
* @param transferId
* @return the trabsfer progress
* @throws TransferException
*/
TransferProgress getStatus(String transferId) throws TransferException;
/**
* Get the version that we are transfering to.
*/
TransferVersion getVersion();
/**
*
*
* @return
*/
TransferProgressMonitor getProgressMonitor();
/**
* get the transfer report for the specified transfer
* @param transferId
*/
InputStream getTransferReport(String transferId);
/**
* set the root node for the file system receiver
* @param rootFileSystem
*/
void setFileTransferRootNodeFileFileSystem(String rootFileSystem);
}

16
source/java/org/alfresco/service/cmr/transfer/TransferVersion.java
View File

@@ -19,33 +19,33 @@
package org.alfresco.service.cmr.transfer;
public interface TransferVersion
{
{
/**
* Gets the major version number, e.g. <u>1</u>.2.3
*
*
* @return major version number
*/
public String getVersionMajor();
/**
* Gets the minor version number, e.g. 1.<u>2</u>.3
*
*
* @return minor version number
*/
public String getVersionMinor();
/**
* Gets the version revision number, e.g. 1.2.<u>3</u>
*
*
* @return revision number
*/
public String getVersionRevision();
/**
* Gets the edition
*
*
* @return the edition
*/
public String getEdition();
}