[Fix #467] Adding schema validation

impl/pom.xml

@@ -26,6 +26,10 @@
       <artifactId>jersey-media-json-jackson</artifactId>
       <version>${version.org.glassfish.jersey}</version>
     </dependency>
+   <dependency>
+     <groupId>com.networknt</groupId>
+     <artifactId>json-schema-validator</artifactId>
+   </dependency>
     <dependency>
       <groupId>net.thisptr</groupId>
       <artifactId>jackson-jq</artifactId>

impl/src/main/java/io/serverlessworkflow/impl/WorkflowContext.java

@@ -19,7 +19,6 @@
 import io.serverlessworkflow.impl.json.JsonUtils;
 
 public class WorkflowContext {
-
   private final WorkflowPosition position;
   private JsonNode context;
   private final JsonNode input;

impl/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java

@@ -15,44 +15,79 @@
  */
 package io.serverlessworkflow.impl;
 
+import static io.serverlessworkflow.impl.WorkflowUtils.*;
 import static io.serverlessworkflow.impl.json.JsonUtils.*;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.Input;
+import io.serverlessworkflow.api.types.Output;
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.executors.DefaultTaskExecutorFactory;
 import io.serverlessworkflow.impl.executors.TaskExecutor;
 import io.serverlessworkflow.impl.executors.TaskExecutorFactory;
+import io.serverlessworkflow.impl.expressions.ExpressionFactory;
+import io.serverlessworkflow.impl.expressions.JQExpressionFactory;
 import io.serverlessworkflow.impl.json.JsonUtils;
+import io.serverlessworkflow.impl.jsonschema.DefaultSchemaValidatorFactory;
+import io.serverlessworkflow.impl.jsonschema.SchemaValidator;
+import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
+import io.serverlessworkflow.resources.DefaultResourceLoaderFactory;
+import io.serverlessworkflow.resources.ResourceLoaderFactory;
+import java.nio.file.Path;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class WorkflowDefinition {
 
   private WorkflowDefinition(
       Workflow workflow,
-      TaskExecutorFactory taskFactory,
-      Collection<WorkflowExecutionListener> listeners) {
+      Collection<WorkflowExecutionListener> listeners,
+      WorkflowFactories factories) {
     this.workflow = workflow;
-    this.taskFactory = taskFactory;
     this.listeners = listeners;
+    this.factories = factories;
+    if (workflow.getInput() != null) {
+      Input input = workflow.getInput();
+      this.inputSchemaValidator =
+          getSchemaValidator(
+              factories.getValidatorFactory(), schemaToNode(factories, input.getSchema()));
+      this.inputFilter = buildWorkflowFilter(factories.getExpressionFactory(), input.getFrom());
+    }
+    if (workflow.getOutput() != null) {
+      Output output = workflow.getOutput();
+      this.outputSchemaValidator =
+          getSchemaValidator(
+              factories.getValidatorFactory(), schemaToNode(factories, output.getSchema()));
+      this.outputFilter = buildWorkflowFilter(factories.getExpressionFactory(), output.getAs());
+    }
   }
 
   private final Workflow workflow;
   private final Collection<WorkflowExecutionListener> listeners;
-  private final TaskExecutorFactory taskFactory;
+  private final WorkflowFactories factories;
+  private Optional<SchemaValidator> inputSchemaValidator = Optional.empty();
+  private Optional<SchemaValidator> outputSchemaValidator = Optional.empty();
+  private Optional<WorkflowFilter> inputFilter = Optional.empty();
+  private Optional<WorkflowFilter> outputFilter = Optional.empty();
+
   private final Map<String, TaskExecutor<? extends TaskBase>> taskExecutors =
       new ConcurrentHashMap<>();
 
   public static class Builder {
     private final Workflow workflow;
     private TaskExecutorFactory taskFactory = DefaultTaskExecutorFactory.get();
+    private ExpressionFactory exprFactory = JQExpressionFactory.get();
     private Collection<WorkflowExecutionListener> listeners;
+    private ResourceLoaderFactory resourceLoaderFactory = DefaultResourceLoaderFactory.get();
+    private SchemaValidatorFactory schemaValidatorFactory = DefaultSchemaValidatorFactory.get();
+    private Path path;
 
     private Builder(Workflow workflow) {
       this.workflow = workflow;
@@ -71,13 +106,39 @@ public Builder withTaskExecutorFactory(TaskExecutorFactory factory) {
       return this;
     }
 
+    public Builder withExpressionFactory(ExpressionFactory factory) {
+      this.exprFactory = factory;
+      return this;
+    }
+
+    public Builder withPath(Path path) {
+      this.path = path;
+      return this;
+    }
+
+    public Builder withResourceLoaderFactory(ResourceLoaderFactory resourceLoader) {
+      this.resourceLoaderFactory = resourceLoader;
+      return this;
+    }
+
+    public Builder withSchemaValidatorFactory(SchemaValidatorFactory factory) {
+      this.schemaValidatorFactory = factory;
+      return this;
+    }
+
     public WorkflowDefinition build() {
-      return new WorkflowDefinition(
-          workflow,
-          taskFactory,
-          listeners == null
-              ? Collections.emptySet()
-              : Collections.unmodifiableCollection(listeners));
+      WorkflowDefinition def =
+          new WorkflowDefinition(
+              workflow,
+              listeners == null
+                  ? Collections.emptySet()
+                  : Collections.unmodifiableCollection(listeners),
+              new WorkflowFactories(
+                  taskFactory,
+                  resourceLoaderFactory.getResourceLoader(path),
+                  exprFactory,
+                  schemaValidatorFactory));
+      return def;
     }
   }
 
@@ -86,7 +147,7 @@ public static Builder builder(Workflow workflow) {
   }
 
   public WorkflowInstance execute(Object input) {
-    return new WorkflowInstance(taskFactory, JsonUtils.fromValue(input));
+    return new WorkflowInstance(JsonUtils.fromValue(input));
   }
 
   enum State {
@@ -101,11 +162,15 @@ public class WorkflowInstance {
     private State state;
     private WorkflowContext context;
 
-    private WorkflowInstance(TaskExecutorFactory factory, JsonNode input) {
+    private WorkflowInstance(JsonNode input) {
       this.output = input;
-      this.state = State.STARTED;
+      inputSchemaValidator.ifPresent(v -> v.validate(input));
       this.context = WorkflowContext.builder(input).build();
+      inputFilter.ifPresent(f -> output = f.apply(context, Optional.empty(), output));
+      this.state = State.STARTED;
       processDo(workflow.getDo());
+      outputFilter.ifPresent(f -> output = f.apply(context, Optional.empty(), output));
+      outputSchemaValidator.ifPresent(v -> v.validate(output));
     }
 
     private void processDo(List<TaskItem> tasks) {
@@ -118,7 +183,7 @@ private void processDo(List<TaskItem> tasks) {
             taskExecutors
                 .computeIfAbsent(
                     context.position().jsonPointer(),
-                    k -> taskFactory.getTaskExecutor(task.getTask()))
+                    k -> factories.getTaskFactory().getTaskExecutor(task.getTask(), factories))
                 .apply(context, output);
         listeners.forEach(l -> l.onTaskEnded(context.position(), task.getTask()));
         context.position().back().back();

impl/src/main/java/io/serverlessworkflow/impl/WorkflowFactories.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * 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.
+ */
+package io.serverlessworkflow.impl;
+
+import io.serverlessworkflow.impl.executors.TaskExecutorFactory;
+import io.serverlessworkflow.impl.expressions.ExpressionFactory;
+import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
+import io.serverlessworkflow.resources.ResourceLoader;
+
+public class WorkflowFactories {
+
+  private final TaskExecutorFactory taskFactory;
+  private final ResourceLoader resourceLoader;
+  private final ExpressionFactory expressionFactory;
+  private final SchemaValidatorFactory validatorFactory;
+
+  public WorkflowFactories(
+      TaskExecutorFactory taskFactory,
+      ResourceLoader resourceLoader,
+      ExpressionFactory expressionFactory,
+      SchemaValidatorFactory validatorFactory) {
+    this.taskFactory = taskFactory;
+    this.resourceLoader = resourceLoader;
+    this.expressionFactory = expressionFactory;
+    this.validatorFactory = validatorFactory;
+  }
+
+  public TaskExecutorFactory getTaskFactory() {
+    return taskFactory;
+  }
+
+  public ResourceLoader getResourceLoader() {
+    return resourceLoader;
+  }
+
+  public ExpressionFactory getExpressionFactory() {
+    return expressionFactory;
+  }
+
+  public SchemaValidatorFactory getValidatorFactory() {
+    return validatorFactory;
+  }
+}

impl/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * 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.
+ */
+package io.serverlessworkflow.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Optional;
+
+@FunctionalInterface
+public interface WorkflowFilter {
+  JsonNode apply(WorkflowContext workflow, Optional<TaskContext<?>> task, JsonNode node);
+}

impl/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * 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.
+ */
+package io.serverlessworkflow.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.serverlessworkflow.api.WorkflowFormat;
+import io.serverlessworkflow.api.types.ExportAs;
+import io.serverlessworkflow.api.types.InputFrom;
+import io.serverlessworkflow.api.types.OutputAs;
+import io.serverlessworkflow.api.types.SchemaExternal;
+import io.serverlessworkflow.api.types.SchemaInline;
+import io.serverlessworkflow.api.types.SchemaUnion;
+import io.serverlessworkflow.impl.expressions.Expression;
+import io.serverlessworkflow.impl.expressions.ExpressionFactory;
+import io.serverlessworkflow.impl.expressions.ExpressionUtils;
+import io.serverlessworkflow.impl.json.JsonUtils;
+import io.serverlessworkflow.impl.jsonschema.SchemaValidator;
+import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
+import io.serverlessworkflow.resources.StaticResource;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import java.util.Optional;
+
+public class WorkflowUtils {
+
+  private WorkflowUtils() {}
+
+  public static Optional<SchemaValidator> getSchemaValidator(
+      SchemaValidatorFactory validatorFactory, Optional<JsonNode> node) {
+    return node.map(n -> validatorFactory.getValidator(n));
+  }
+
+  public static Optional<JsonNode> schemaToNode(WorkflowFactories factories, SchemaUnion schema) {
+    if (schema != null) {
+      if (schema.getSchemaInline() != null) {
+        SchemaInline inline = schema.getSchemaInline();
+        return Optional.of(JsonUtils.mapper().convertValue(inline.getDocument(), JsonNode.class));
+      } else if (schema.getSchemaExternal() != null) {
+        SchemaExternal external = schema.getSchemaExternal();
+        StaticResource resource = factories.getResourceLoader().loadStatic(external.getResource());
+        ObjectMapper mapper = WorkflowFormat.fromFileName(resource.name()).mapper();
+        try (InputStream in = resource.open()) {
+          return Optional.of(mapper.readTree(in));
+        } catch (IOException io) {
+          throw new UncheckedIOException(io);
+        }
+      }
+    }
+    return Optional.empty();
+  }
+
+  public static Optional<WorkflowFilter> buildWorkflowFilter(
+      ExpressionFactory exprFactory, InputFrom from) {
+    return from != null
+        ? Optional.of(buildWorkflowFilter(exprFactory, from.getString(), from.getObject()))
+        : Optional.empty();
+  }
+
+  public static Optional<WorkflowFilter> buildWorkflowFilter(
+      ExpressionFactory exprFactory, OutputAs as) {
+    return as != null
+        ? Optional.of(buildWorkflowFilter(exprFactory, as.getString(), as.getObject()))
+        : Optional.empty();
+  }
+
+  public static Optional<WorkflowFilter> buildWorkflowFilter(
+      ExpressionFactory exprFactory, ExportAs as) {
+    return as != null
+        ? Optional.of(buildWorkflowFilter(exprFactory, as.getString(), as.getObject()))
+        : Optional.empty();
+  }
+
+  private static WorkflowFilter buildWorkflowFilter(
+      ExpressionFactory exprFactory, String str, Object object) {
+    if (str != null) {
+      Expression expression = exprFactory.getExpression(str);
+      return expression::eval;
+    } else {
+      Object exprObj = ExpressionUtils.buildExpressionObject(object, exprFactory);
+      return exprObj instanceof Map
+          ? (w, t, n) ->
+              JsonUtils.fromValue(
+                  ExpressionUtils.evaluateExpressionMap((Map<String, Object>) exprObj, w, t, n))
+          : (w, t, n) -> JsonUtils.fromValue(object);
+    }
+  }
+}