Skip to content

Commit

Permalink
Merge pull request #470 from fjtirado/Fix_#467
Browse files Browse the repository at this point in the history
[Fix #467]  Adding schema validation
  • Loading branch information
fjtirado authored Nov 21, 2024
2 parents 52bcbc9 + 3c09007 commit 11d0050
Show file tree
Hide file tree
Showing 29 changed files with 886 additions and 139 deletions.
4 changes: 4 additions & 0 deletions impl/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import io.serverlessworkflow.impl.json.JsonUtils;

public class WorkflowContext {

private final WorkflowPosition position;
private JsonNode context;
private final JsonNode input;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
}
}

Expand All @@ -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 {
Expand All @@ -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) {
Expand All @@ -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();
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
24 changes: 24 additions & 0 deletions impl/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java
Original file line number Diff line number Diff line change
@@ -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);
}
103 changes: 103 additions & 0 deletions impl/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java
Original file line number Diff line number Diff line change
@@ -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);
}
}
}
Loading

0 comments on commit 11d0050

Please sign in to comment.