Rewrote fileprocessing to not use S3 (#5)

Author: mfateev

README.md

@@ -78,7 +78,7 @@ The steps for configuring and building other samples for Java Cadence Client are
         set AWS_SWF_SAMPLES_CONFIG=<Your SDK Directory>
 
 ## Prerequisite
-  Run Cadence Server
+  Run Cadence Server using Docker Compose
 
     curl -O https://raw.githubusercontent.com/uber/cadence/master/docker/docker-compose.yml
     docker-compose up

src/main/java/com/uber/cadence/samples/fileprocessing/FileProcessingActivities.java

@@ -16,52 +16,34 @@
  */
 package com.uber.cadence.samples.fileprocessing;
 
-import com.uber.cadence.WorkflowService;
 import com.uber.cadence.client.WorkflowClient;
-import com.uber.cadence.samples.common.ConfigHelper;
+
+import java.net.URL;
+
+import static com.uber.cadence.samples.common.SampleConstants.DOMAIN;
 
 /**
- * This is used for launching a Workflow instance of FileProcessingWorkflowExample
+ * This is used for launching a Workflow instance of file processing sample.
  */
 public class FileProcessingStarter {
 
-    static final String WORKFLOW_TASK_LIST = "FileProcessing";
-
-    private static WorkflowService.Iface swfService;
-    private static String domain;
-
     public static void main(String[] args) throws Exception {
-
-        // Load configuration
-        ConfigHelper configHelper = ConfigHelper.createConfig();
-
-        // Create the client for Simple Workflow Service
-        swfService = configHelper.createWorkflowClient();
-        domain = configHelper.getDomain();
-
         // Start Workflow instance
-        String sourceBucketName = configHelper.getValueFromConfig(FileProcessingConfigKeys.WORKFLOW_INPUT_SOURCEBUCKETNAME_KEY);
-        String sourceFilename = configHelper.getValueFromConfig(FileProcessingConfigKeys.WORKFLOW_INPUT_SOURCEFILENAME_KEY);
-        String targetBucketName = configHelper.getValueFromConfig(FileProcessingConfigKeys.WORKFLOW_INPUT_TARGETBUCKETNAME_KEY);
-        String targetFilename = configHelper.getValueFromConfig(FileProcessingConfigKeys.WORKFLOW_INPUT_TARGETFILENAME_KEY);
+        WorkflowClient workflowClient = WorkflowClient.newInstance(DOMAIN);
+        FileProcessingWorkflow workflow = workflowClient.newWorkflowStub(FileProcessingWorkflow.class);
 
-        FileProcessingWorkflow.Arguments workflowArgs = new FileProcessingWorkflow.Arguments();
-        workflowArgs.setSourceBucketName(sourceBucketName);
-        workflowArgs.setSourceFilename(sourceFilename);
-        workflowArgs.setTargetBucketName(targetBucketName);
-        workflowArgs.setTargetFilename(targetFilename);
+        System.out.println("Executing FileProcessingWorkflow");
 
-        WorkflowClient workflowClient = WorkflowClient.newInstance(swfService, domain);
-        FileProcessingWorkflow workflow = workflowClient.newWorkflowStub(FileProcessingWorkflow.class);
+        URL source = new URL("http://www.google.com/");
+        URL destination = new URL("http://dummy");
 
         // This is going to block until the workflow completion.
         // This is rarely used in production. Use the commented code below for async start version.
-        System.out.println("Executing FileProcessingWorkflow");
-        workflow.processFile(workflowArgs);
+        workflow.processFile(source, destination);
+        System.out.println("FileProcessingWorkflow completed");
 
         // Use this code instead of the above blocking call to start workflow asynchronously.
-//        WorkflowExecution workflowExecution = WorkflowClient.asyncStart(workflow::processFile, workflowArgs);
-//
+//        WorkflowExecution workflowExecution = WorkflowClient.asyncStart(workflow::processFile, source, destination);
 //        System.out.println("Started periodic workflow with workflowId=\"" + workflowExecution.getWorkflowId()
 //                + "\" and runId=\"" + workflowExecution.getRunId() + "\"");
 //

src/main/java/com/uber/cadence/samples/fileprocessing/FileProcessingActivitiesZipImpl.java

@@ -16,76 +16,38 @@
  */
 package com.uber.cadence.samples.fileprocessing;
 
-import com.amazonaws.services.s3.AmazonS3;
-import com.uber.cadence.WorkflowService;
-import com.uber.cadence.samples.common.ConfigHelper;
 import com.uber.cadence.worker.Worker;
-import com.uber.cadence.worker.WorkerOptions;
 
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.time.Duration;
+import java.lang.management.ManagementFactory;
+
+import static com.uber.cadence.samples.common.SampleConstants.DOMAIN;
 
 /**
- * This is the process which hosts all workflows and activities in this sample
+ * This is the process which hosts all workflows and activities in this sample.
+ * Run multiple instances of the worker in different windows. Then start workflow
+ * by running FileProcessingStarter. Note that all activities always execute on the same worker.
+ * But each time they might end up on a different worker as the first activity is dispatched to the common task list.
  */
 public class FileProcessingWorker {
 
     static final String TASK_LIST = "FileProcessing";
 
-    public static void main(String[] args) throws Exception {
-        ConfigHelper configHelper = ConfigHelper.createConfig();
-        WorkflowService.Iface swfService = configHelper.createWorkflowClient();
-        AmazonS3 s3Client = configHelper.createS3Client();
-        String domain = configHelper.getDomain();
+    public static void main(String[] args) {
 
-        String localFolder = configHelper.getValueFromConfig(FileProcessingConfigKeys.ACTIVITY_WORKER_LOCALFOLDER);
+        String hostSpecifiTaskList = ManagementFactory.getRuntimeMXBean().getName();
 
         // Start worker to poll the common task list
-        final Worker workerForCommonTaskList = new Worker(swfService, domain, TASK_LIST, null);
-        SimpleStoreActivitiesS3Impl storeActivityImpl = new SimpleStoreActivitiesS3Impl(s3Client, localFolder, getHostName());
+        final Worker workerForCommonTaskList = new Worker(DOMAIN, TASK_LIST);
+        workerForCommonTaskList.registerWorkflowImplementationTypes(FileProcessingWorkflowImpl.class);
+        StoreActivitiesImpl storeActivityImpl = new StoreActivitiesImpl(hostSpecifiTaskList);
         workerForCommonTaskList.registerActivitiesImplementations(storeActivityImpl);
-        workerForCommonTaskList.registerWorkflowImplementationTypes(FileProcessingWorkflowZipImpl.class);
-
         workerForCommonTaskList.start();
-        System.out.println("Worker tarted for task list: " + TASK_LIST);
+        System.out.println("Worker started for task list: " + TASK_LIST);
 
         // Start worker to poll the host specific task list
-        WorkerOptions hostSpecificOptions = new WorkerOptions.Builder().build();
-        final Worker workerForHostSpecificTaskList = new Worker(swfService, domain, getHostName(), hostSpecificOptions);
-        FileProcessingActivitiesZipImpl processorActivityImpl = new FileProcessingActivitiesZipImpl(localFolder);
-        workerForHostSpecificTaskList.registerActivitiesImplementations(storeActivityImpl, processorActivityImpl);
+        final Worker workerForHostSpecificTaskList = new Worker(DOMAIN, hostSpecifiTaskList);
+        workerForHostSpecificTaskList.registerActivitiesImplementations(storeActivityImpl);
         workerForHostSpecificTaskList.start();
-        System.out.println("Worker Started for activity and workflow task List: " + getHostName());
-
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-
-            public void run() {
-                workerForCommonTaskList.shutdown(Duration.ofSeconds(5));
-                workerForHostSpecificTaskList.shutdown(Duration.ofSeconds(5));
-                System.out.println("Activity Workers Exited.");
-            }
-        });
-
-        System.out.println("Please press any key to terminate service.");
-
-        try {
-            System.in.read();
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
-        System.exit(0);
-
+        System.out.println("Worker Started for activity task List: " + hostSpecifiTaskList);
     }
-
-    static String getHostName() {
-        try {
-            InetAddress addr = InetAddress.getLocalHost();
-            return addr.getHostName();
-        } catch (UnknownHostException e) {
-            throw new Error(e);
-        }
-    }
-
 }

src/main/java/com/uber/cadence/samples/fileprocessing/FileProcessingConfigKeys.java

@@ -18,53 +18,14 @@
 
 import com.uber.cadence.workflow.WorkflowMethod;
 
+import java.net.URL;
+
 /**
  * Contract for file processing workflow
  */
 public interface FileProcessingWorkflow {
 
-    final class Arguments {
-        String sourceBucketName;
-        String sourceFilename;
-        String targetBucketName;
-        String targetFilename;
-
-        public String getSourceBucketName() {
-            return sourceBucketName;
-        }
-
-        public void setSourceBucketName(String sourceBucketName) {
-            this.sourceBucketName = sourceBucketName;
-        }
-
-        public String getSourceFilename() {
-            return sourceFilename;
-        }
-
-        public void setSourceFilename(String sourceFilename) {
-            this.sourceFilename = sourceFilename;
-        }
-
-        public String getTargetBucketName() {
-            return targetBucketName;
-        }
-
-        public void setTargetBucketName(String targetBucketName) {
-            this.targetBucketName = targetBucketName;
-        }
-
-        public String getTargetFilename() {
-            return targetFilename;
-        }
-
-        public void setTargetFilename(String targetFilename) {
-            this.targetFilename = targetFilename;
-        }
-    }
-    /**
-     * Uses a structure as arguments, to make addition of new arguments a backwards compatible change.
-     */
-    @WorkflowMethod(taskList = FileProcessingStarter.WORKFLOW_TASK_LIST,
-            executionStartToCloseTimeoutSeconds = 300)
-    void processFile(Arguments args);
+    @WorkflowMethod(taskList = FileProcessingWorker.TASK_LIST,
+            executionStartToCloseTimeoutSeconds = 30)
+    void processFile(URL source, URL destination);
 }