Property to include SingleTable library in template in Java Processors

dataspray-core/src/main/java/io/dataspray/core/BuilderImpl.java

@@ -131,7 +131,7 @@ private Path getArtifactPath(Project project, Processor processor) {
         Path codeZipPath;
         if (processor instanceof JavaProcessor) {
             codeZipPath = project.getProcessorDir(processor)
-                    .resolve(Path.of("target", processor.getTaskId() + ".jar"));
+                    .resolve(Path.of("target", processor.getProcessorId() + ".jar"));
         } else if (processor instanceof TypescriptProcessor) {
             codeZipPath = project.getProcessorDir(processor)
                     .resolve(Path.of("dist", "index.zip"));

dataspray-core/src/main/java/io/dataspray/core/StreamRuntimeImpl.java

@@ -119,7 +119,7 @@ public String upload(Organization organization, Project project, String processo
         UploadCodeResponse uploadCodeResponse = DataSprayClient.get(organization.toAccess())
                 .control()
                 .uploadCode(organization.getName(), new UploadCodeRequest()
-                        .taskId(processor.getTaskId())
+                        .taskId(processor.getProcessorId())
                         .contentLengthBytes(codeZipFile.length()));
 
         // Upload to S3
@@ -161,7 +161,7 @@ public String publish(Organization organization, Project project, String process
                 handler);
         TaskVersion deployedVersion = DataSprayClient.get(organization.toAccess())
                 .control()
-                .deployVersion(organization.getName(), processor.getTaskId(), new DeployRequest()
+                .deployVersion(organization.getName(), processor.getProcessorId(), new DeployRequest()
                         .runtime(runtime)
                         .handler(handler)
                         .inputQueueNames(processor.getInputStreams().stream()
@@ -206,7 +206,7 @@ public TaskStatus activateVersion(Organization organization, Project project, St
         log.info("Activating version {} for task {}", version, processorName);
         TaskStatus taskStatus = DataSprayClient.get(organization.toAccess())
                 .control()
-                .activateVersion(organization.getName(), processor.getTaskId(), version);
+                .activateVersion(organization.getName(), processor.getProcessorId(), version);
         log.info("Version active!");
 
         return taskStatus;
@@ -219,10 +219,10 @@ public TaskStatus pause(Organization organization, Project project, String proce
         checkState(Processor.Target.DATASPRAY.equals(processor.getTarget()),
                 "Not yet implemented: %s", processor.getTarget());
 
-        log.info("Pausing {}", processor.getTaskId());
+        log.info("Pausing {}", processor.getProcessorId());
         TaskStatus taskStatus = DataSprayClient.get(organization.toAccess())
                 .control()
-                .pause(organization.getName(), processor.getTaskId());
+                .pause(organization.getName(), processor.getProcessorId());
         log.info("Task set to be paused");
         printStatus(taskStatus);
 
@@ -236,10 +236,10 @@ public TaskStatus resume(Organization organization, Project project, String proc
         checkState(Processor.Target.DATASPRAY.equals(processor.getTarget()),
                 "Not yet implemented: %s", processor.getTarget());
 
-        log.info("Resuming {}", processor.getTaskId());
+        log.info("Resuming {}", processor.getProcessorId());
         TaskStatus taskStatus = DataSprayClient.get(organization.toAccess())
                 .control()
-                .resume(organization.getName(), processor.getTaskId());
+                .resume(organization.getName(), processor.getProcessorId());
         log.info("Task set to be resumed");
         printStatus(taskStatus);
 
@@ -255,7 +255,7 @@ public TaskVersions listVersions(Organization organization, Project project, Str
 
         TaskVersions versions = DataSprayClient.get(organization.toAccess())
                 .control()
-                .getVersions(organization.getName(), processor.getTaskId());
+                .getVersions(organization.getName(), processor.getProcessorId());
         printVersions(versions);
 
         return versions;
@@ -270,7 +270,7 @@ public TaskStatus delete(Organization organization, Project project, String proc
 
         TaskStatus status = DataSprayClient.get(organization.toAccess())
                 .control()
-                .delete(organization.getName(), processor.getTaskId());
+                .delete(organization.getName(), processor.getProcessorId());
 
         printStatus(status);
 

dataspray-core/src/main/java/io/dataspray/core/definition/model/Item.java

@@ -51,10 +51,6 @@ public String getNameUpper() {
         return name.toUpperCase();
     }
 
-    public String getTaskId() {
-        return getNameDir();
-    }
-
     @Cacheable(lifetime = Definition.CACHEABLE_METHODS_LIFETIME_IN_MIN)
     public String getNameCamelUpper() {
         return StringUtil.camelCase(name, true);

dataspray-core/src/main/java/io/dataspray/core/definition/model/JavaProcessor.java

@@ -37,6 +37,8 @@
 @RegisterForReflection
 public class JavaProcessor extends Processor {
 
+    Boolean includeSingleTableLibrary;
+
     @Override
     public void initialize() {
         super.initialize();

dataspray-core/src/main/java/io/dataspray/core/definition/model/Processor.java

@@ -45,6 +45,11 @@
 @NonFinal
 @RegisterForReflection
 public class Processor extends Item {
+
+    public String getProcessorId() {
+        return getNameDir();
+    }
+
     @Nonnull
     Target target;
 

dataspray-core/src/main/resources/template/java/pom-parent.xml.template.mustache

@@ -53,7 +53,7 @@
         <dependency>
             <groupId>io.dataspray</groupId>
             <artifactId>dataspray-runner</artifactId>
-            <version>0.0.8</version>
+            <version>0.0.9</version>
         </dependency>
         <dependency>
             <groupId>com.amazonaws</groupId>
@@ -65,6 +65,15 @@
             <artifactId>aws-lambda-java-events</artifactId>
             <version>3.11.1</version>
         </dependency>
+        {{#processor.hasDynamoState}}
+        {{#processor.includeSingleTableLibrary}}
+        <dependency>
+            <groupId>io.dataspray</groupId>
+            <artifactId>single-table</artifactId>
+            <version>2.2.5</version>
+        </dependency>
+        {{/processor.includeSingleTableLibrary}}
+        {{/processor.hasDynamoState}}
         {{^processor.jsonDataFormats.empty}}
         <dependency>
             <groupId>com.google.code.gson</groupId>
@@ -143,7 +152,7 @@
             <plugin>
                 <groupId>org.jsonschema2pojo</groupId>
                 <artifactId>jsonschema2pojo-maven-plugin</artifactId>
-                <version>1.1.2</version>
+                <version>1.2.2</version>
                 <configuration>
                     <sourcePaths>
                         {{#processor.jsonDataFormats}}
@@ -173,6 +182,10 @@
                     <dateType>java.time.LocalDate</dateType>
                     <timeType>java.time.LocalTime</timeType>
                     <includeGeneratedAnnotation>false</includeGeneratedAnnotation>
+                    <useJakartaValidation>true</useJakartaValidation>
+                    <initializeCollections>true</initializeCollections>
+                    <includeConstructors>true</includeConstructors>
+                    <includeRequiredPropertiesConstructor>true</includeRequiredPropertiesConstructor>
                 </configuration>
                 <executions>
                     <execution>

dataspray-core/src/main/resources/template/java/src/main/java/{{{definition.javaPackagePath}}}/autogen/CoordinatorImpl.java.template.mustache

@@ -10,6 +10,7 @@ import java.nio.charset.StandardCharsets;
 import java.io.ByteArrayOutputStream;
 {{#processor.hasDynamoState}}
 import io.dataspray.runner.StateManager;
+import io.dataspray.runner.StateManagerFactoryImpl;
 import java.time.Duration;
 import java.util.Optional;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
@@ -37,14 +38,14 @@ public class CoordinatorImpl implements {{#processor.web}}WebCoordinator{{#proce
 
     @Override
     public StateManager stateForMessageKey(Optional<Duration> ttl) {
-        return stateForNamespace(ttl, "task", "{{{processor.taskId}}}", "key", messageKey
+        return stateForNamespace(ttl, "task", "{{{processor.processorId}}}", "key", messageKey
             .orElseThrow(() -> new IllegalStateException("messageKey is not set")));
     }
     {{/processor.hasInputStreams}}
 
     @Override
     public StateManager stateForTask(Optional<Duration> ttl) {
-        return stateForNamespace(ttl, "task", "{{{processor.taskId}}}");
+        return stateForNamespace(ttl, "task", "{{{processor.processorId}}}");
     }
 
     @Override
@@ -56,6 +57,16 @@ public class CoordinatorImpl implements {{#processor.web}}WebCoordinator{{#proce
     public DynamoDbClient getDynamoClient() {
         return rawCoordinator.getDynamoClient();
     }
+    {{#processor.includeSingleTableLibrary}}
+
+    @Override
+    public SingleTable getSingleTable() {
+        return SingleTable.builder()
+            .tableName(System.getenv(StateManagerFactoryImpl.DATASPRAY_STATE_TABLE_NAME_ENV))
+            .overrideGson(GsonUtil.get())
+            .build()
+    }
+    {{/processor.includeSingleTableLibrary}}
     {{/processor.hasDynamoState}}
     {{#processor.outputStreams}}
 

dataspray-core/src/main/resources/template/java/src/main/java/{{{definition.javaPackagePath}}}/autogen/StreamCoordinator.java.template.mustache

@@ -48,6 +48,13 @@ public interface StreamCoordinator {
      * Use this method to perform advanced operations on DynamoDB.
      */
     DynamoDbClient getDynamoClient();
+    {{#processor.includeSingleTableLibrary}}
+
+    /**
+     * Returns an instance of SingleTable which is a convenience wrapper on top of DynamoDB client.
+     */
+    SingleTable getSingleTable();
+    {{/processor.includeSingleTableLibrary}}
     {{/processor.hasDynamoState}}
     {{#processor.outputStreams}}
 

dataspray-core/src/main/resources/template/java/src/main/java/{{{definition.javaPackagePath}}}/autogen/WebCoordinator.java.template.mustache

@@ -39,6 +39,13 @@ public interface WebCoordinator {
      * Use this method to perform advanced operations on DynamoDB.
      */
     DynamoDbClient getDynamoClient();
+    {{#processor.includeSingleTableLibrary}}
+
+    /**
+     * Returns an instance of SingleTable which is a convenience wrapper on top of DynamoDB client.
+     */
+    SingleTable getSingleTable();
+    {{/processor.includeSingleTableLibrary}}
     {{/processor.hasDynamoState}}
     {{#processor.outputStreams}}
 

dataspray-core/src/main/resources/template/java/src/test/java/{{{definition.javaPackagePath}}}/autogen/TestCoordinator.java.template.mustache

@@ -55,13 +55,13 @@ public class TestCoordinator implements {{#processor.web}}WebCoordinator{{#proce
     @Override
     public StateManager stateForMessageKey(Optional<Duration> ttl) {
         checkState(messageKey != null, "messageKey is not set");
-        return stateForNamespace(ttl, "task", "{{{processor.taskId}}}", "key", messageKey);
+        return stateForNamespace(ttl, "task", "{{{processor.processorId}}}", "key", messageKey);
     }
     {{/processor.hasInputStreams}}
 
     @Override
     public StateManager stateForTask(Optional<Duration> ttl) {
-        return stateForNamespace(ttl, "task", "{{{processor.taskId}}}");
+        return stateForNamespace(ttl, "task", "{{{processor.processorId}}}");
     }
 
     @Override

dataspray-core/src/main/resources/template/root/ARCHITECTURE.md.template.mustache

@@ -3,19 +3,19 @@
 ```mermaid
 graph TB;
 {{#definition.processors}}
-    {{taskId}}[{{name}}];
+    {{processorId}}[{{name}}];
     {{#inputStreams}}
-        Queue{{{uniqueNameCamelUpper}}}([Queue {{{uniqueNameCamelUpper}}}]) --> {{taskId}};
+        Queue{{{uniqueNameCamelUpper}}}([Queue {{{uniqueNameCamelUpper}}}]) --> {{processorId}};
     {{/inputStreams}}
     {{#outputStreams}}
-        {{taskId}} --> Queue{{{uniqueNameCamelUpper}}}([Queue {{{uniqueNameCamelUpper}}}]);
+        {{processorId}} --> Queue{{{uniqueNameCamelUpper}}}([Queue {{{uniqueNameCamelUpper}}}]);
     {{/outputStreams}}
     {{#web}}
     {{#endpoints}}
-        Web{{{nameCamelUpper}}}([Web {{{nameCamelUpper}}}]) --> {{taskId}};
+        Web{{{nameCamelUpper}}}([Web {{{nameCamelUpper}}}]) --> {{processorId}};
     {{/endpoints}}
     {{#endpoints.empty}}
-        Web{{{nameCamelUpper}}}([Web]) --> {{taskId}};
+        Web{{{nameCamelUpper}}}([Web]) --> {{processorId}};
     {{/endpoints.empty}}
     {{/web}}
 {{/definition.processors}}

dataspray-core/src/main/resources/template/typescript/src/autogen/coordinator.ts.template.mustache

@@ -132,11 +132,11 @@ export class CoordinatorImpl implements {{#processor.web}}WebCoordinator{{#proce
         if (this.messageKey === undefined) {
             throw new Error('messageKey is not set');
         }
-        return this.stateForNamespace(ttlInSec, ["task", "{{{processor.taskId}}}", "key", this.messageKey]);
+        return this.stateForNamespace(ttlInSec, ["task", "{{{processor.processorId}}}", "key", this.messageKey]);
     }
 
     stateForTask(ttlInSec?: number): StateManager {
-        return this.stateForNamespace(ttlInSec, ["task", "{{{processor.taskId}}}"]);
+        return this.stateForNamespace(ttlInSec, ["task", "{{{processor.processorId}}}"]);
     }
 
     stateForNamespace(ttlInSec: number | undefined, namespace: string[]): StateManager {