Add a webdav archive worker

Author: bartv

archive/pom.xml

@@ -12,4 +12,11 @@
 	<groupId>taskworker.workers</groupId>
 	<artifactId>archive</artifactId>
 
+	<dependencies>
+		<dependency>
+		  <groupId>com.github.lookfirst</groupId>
+		  <artifactId>sardine</artifactId>
+		  <version>5.0.1</version>
+		</dependency>
+	</dependencies>
 </project>

archive/src/main/java/drm/taskworker/workers/WebdavArchiveWorker.java

@@ -0,0 +1,91 @@
+/*
+    Copyright 2013 KU Leuven Research and Development - iMinds - Distrinet
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+    Administrative Contact: [email protected]
+    Technical Contact: [email protected]
+*/
+
+package drm.taskworker.workers;
+
+import static drm.taskworker.config.Config.cfg;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.logging.Level;
+
+import com.github.sardine.DavResource;
+import com.github.sardine.Sardine;
+import com.github.sardine.SardineFactory;
+
+import drm.taskworker.Worker;
+import drm.taskworker.tasks.ParameterFoundException;
+import drm.taskworker.tasks.Task;
+import drm.taskworker.tasks.TaskResult;
+
+/**
+ * Collect all files in a workflow and zip them when an end of workflow task
+ * is received.
+ * 
+ * @author Bart Vanbrabant <[email protected]>
+ */
+public class WebdavArchiveWorker extends Worker {
+	/**
+	 * Creates a new work with the name blob-to-cache
+	 */
+	public WebdavArchiveWorker(String workerName) {
+		super(workerName);
+	}
+
+	/**
+	 * Archive the result of the previous task
+	 */
+	public TaskResult work(Task task) {
+		logger.info("Archiving file");
+		TaskResult result = new TaskResult();
+		
+		byte[] fileData = null;
+		
+		try {
+			fileData = (byte[])task.getParam("arg0");
+		} catch (ParameterFoundException e) {
+			return result.setResult(TaskResult.Result.ARGUMENT_ERROR);
+		}
+		
+		try {
+			// WebDAV URL:
+			final String baseUrl = cfg().getProperty("taskworker.archive.url");
+			if (baseUrl == null) {
+				logger.log(Level.SEVERE, "No base url configured to upload result to.");
+				return result.setResult(TaskResult.Result.ARGUMENT_ERROR);
+			}
+			
+			Sardine sardine = SardineFactory.begin(cfg().getProperty("taskworker.archive.username", "admin"), 
+					cfg().getProperty("taskworker.archive.password", "admin"));
+			List<DavResource> resources = sardine.list(baseUrl);
+			
+			sardine.put(baseUrl + task.getJobId().toString(), fileData, "application/zip");
+
+			Task newTask = new Task(task, this.getNextWorker(task.getJobId()));
+			result.addNextTask(newTask);
+			
+		} catch (IOException e) {
+			result.setResult(TaskResult.Result.EXCEPTION);
+			result.setException(e);
+			return result;
+		}
+
+		return result.setResult(TaskResult.Result.SUCCESS);
+	}
+}

join/src/main/java/drm/taskworker/workers/JoinWorker.java

@@ -86,8 +86,10 @@ public TaskResult work(Task task) {
 
 				// load all parents and build the map of parameters
 				Map<String, List<Object>> varMap = new HashMap<>();
+				List<Task> parents = newTask.getParents();
+				logger.info("Joining the results of " + parents.size() + " parents");
 
-				for (Task parentTask : newTask.getParents()) {
+				for (Task parentTask : parents) {
 					for (String paramName : parentTask.getParamNames()) {
 						if (!varMap.containsKey(paramName)) {
 							varMap.put(paramName, new ArrayList<Object>());

workflow-upnxt/config.yaml

@@ -1,12 +1,12 @@
 ---
 workers:
-    - {name: csv-to-task, class: drm.taskworker.workers.CSVtoTaskWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: template-to-xml, class: drm.taskworker.workers.TemplateWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: xsl-fo-render, class: drm.taskworker.workers.XslFoRenderWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: zip-files, class: drm.taskworker.workers.ZipWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: join, class: drm.taskworker.workers.JoinWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: archive, class: drm.taskworker.workers.ArchiveWorker, threads: 2, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
-    - {name: end, class: drm.taskworker.workers.EndWorker, threads: 1, code: "target/upnxt-0.2.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: csv-to-task, class: drm.taskworker.workers.CSVtoTaskWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: template-to-xml, class: drm.taskworker.workers.TemplateWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: xsl-fo-render, class: drm.taskworker.workers.XslFoRenderWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: zip-files, class: drm.taskworker.workers.ZipWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: join, class: drm.taskworker.workers.JoinWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: archive, class: drm.taskworker.workers.WebdavArchiveWorker, threads: 2, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
+    - {name: end, class: drm.taskworker.workers.EndWorker, threads: 1, code: "target/upnxt-0.4.0-SNAPSHOT-jar-with-dependencies.jar"}
 
 
 workflows:
@@ -20,10 +20,6 @@ workflows:
             archive: {next: end}
         start: csv-to-task
         end: end
-        timeout: {
-          default: 15,
-          xsl-fo-render: 30
-        }
 
 scheduler:
     class: drm.taskworker.schedule.FairShare