diff --git a/impl/core/src/main/java/org/openworkflow/sdk/impl/LifecycleEvents.java b/impl/core/src/main/java/org/openworkflow/sdk/impl/LifecycleEvents.java index b74da2379..b7fd18685 100644 --- a/impl/core/src/main/java/org/openworkflow/sdk/impl/LifecycleEvents.java +++ b/impl/core/src/main/java/org/openworkflow/sdk/impl/LifecycleEvents.java @@ -15,6 +15,9 @@ */ package org.openworkflow.sdk.impl; +import org.openworkflow.sdk.impl.lifecycle.EventType; +import org.openworkflow.sdk.impl.lifecycle.LifeCycleEventType; + /** * @see Open @@ -24,49 +27,50 @@ public final class LifecycleEvents { private LifecycleEvents() {} - private static final String SPEC_PREFIX = "org.open-workflow-specification."; + private static final String SPEC_PREFIX = LifeCycleEventType.V8.prefix(); /** Notifies about the start of a task. */ - public static final String TASK_STARTED = SPEC_PREFIX + "task.started.v1"; + public static final String TASK_STARTED = SPEC_PREFIX + EventType.TASK_STARTED.type(); /** Notifies about the completion of a task's execution. */ - public static final String TASK_COMPLETED = SPEC_PREFIX + "task.completed.v1"; + public static final String TASK_COMPLETED = SPEC_PREFIX + EventType.TASK_COMPLETED.type(); /** Notifies about suspending a task's execution. */ - public static final String TASK_SUSPENDED = SPEC_PREFIX + "task.suspended.v1"; + public static final String TASK_SUSPENDED = SPEC_PREFIX + EventType.TASK_SUSPENDED.type(); /** Notifies about resuming a task's execution. */ - public static final String TASK_RESUMED = SPEC_PREFIX + "task.resumed.v1"; + public static final String TASK_RESUMED = SPEC_PREFIX + EventType.TASK_RESUMED.type(); /** Notifies about a task being faulted. */ - public static final String TASK_FAULTED = SPEC_PREFIX + "task.faulted.v1"; + public static final String TASK_FAULTED = SPEC_PREFIX + EventType.TASK_FAULTED.type(); /** Notifies about the cancellation of a task's execution. */ - public static final String TASK_CANCELLED = SPEC_PREFIX + "task.cancelled.v1"; + public static final String TASK_CANCELLED = SPEC_PREFIX + EventType.TASK_CANCELLED.type(); /** Notifies about retrying a task's execution. */ - public static final String TASK_RETRIED = SPEC_PREFIX + "task.retried.v1"; + public static final String TASK_RETRIED = SPEC_PREFIX + EventType.TASK_RETRIED.type(); /** Notifies about the start of a workflow. */ - public static final String WORKFLOW_STARTED = SPEC_PREFIX + "workflow.started.v1"; + public static final String WORKFLOW_STARTED = SPEC_PREFIX + EventType.WORKFLOW_STARTED.type(); /** Notifies about the completion of a workflow execution. */ - public static final String WORKFLOW_COMPLETED = SPEC_PREFIX + "workflow.completed.v1"; + public static final String WORKFLOW_COMPLETED = SPEC_PREFIX + EventType.WORKFLOW_COMPLETED.type(); /** Notifies about suspending a workflow execution. */ - public static final String WORKFLOW_SUSPENDED = SPEC_PREFIX + "workflow.suspended.v1"; + public static final String WORKFLOW_SUSPENDED = SPEC_PREFIX + EventType.WORKFLOW_SUSPENDED.type(); /** Notifies about resuming a workflow execution. */ - public static final String WORKFLOW_RESUMED = SPEC_PREFIX + "workflow.resumed.v1"; + public static final String WORKFLOW_RESUMED = SPEC_PREFIX + EventType.WORKFLOW_RESUMED.type(); /** Notifies about a workflow being faulted. */ - public static final String WORKFLOW_FAULTED = SPEC_PREFIX + "workflow.faulted.v1"; + public static final String WORKFLOW_FAULTED = SPEC_PREFIX + EventType.WORKFLOW_FAULTED.type(); /** Notifies about the cancellation of a workflow execution. */ - public static final String WORKFLOW_CANCELLED = SPEC_PREFIX + "workflow.cancelled.v1"; + public static final String WORKFLOW_CANCELLED = SPEC_PREFIX + EventType.WORKFLOW_CANCELLED.type(); /** Notifies about the change of a workflow's status phase. */ - public static final String WORKFLOW_STATUS_CHANGED = SPEC_PREFIX + "workflow.status-changed.v1"; + public static final String WORKFLOW_STATUS_CHANGED = + SPEC_PREFIX + EventType.WORKFLOW_STATUS_CHANGED.type(); /** * CloudEvent extension attribute carried by every lifecycle event, holding the {@link diff --git a/impl/core/src/main/java/org/openworkflow/sdk/impl/WorkflowApplication.java b/impl/core/src/main/java/org/openworkflow/sdk/impl/WorkflowApplication.java index 4a1152875..55df748ab 100644 --- a/impl/core/src/main/java/org/openworkflow/sdk/impl/WorkflowApplication.java +++ b/impl/core/src/main/java/org/openworkflow/sdk/impl/WorkflowApplication.java @@ -24,6 +24,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.EnumSet; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -61,6 +62,7 @@ import org.openworkflow.sdk.impl.executors.TaskExecutorFactory; import org.openworkflow.sdk.impl.expressions.ExpressionFactory; import org.openworkflow.sdk.impl.expressions.RuntimeDescriptor; +import org.openworkflow.sdk.impl.lifecycle.LifeCycleEventType; import org.openworkflow.sdk.impl.lifecycle.WorkflowExecutionCompletableListener; import org.openworkflow.sdk.impl.lifecycle.WorkflowExecutionListener; import org.openworkflow.sdk.impl.lifecycle.WorkflowExecutionListenerAdapter; @@ -116,6 +118,7 @@ public class WorkflowApplication implements AutoCloseable { private final ScheduledExecutorService schedulerExecutorService; private final Set allowedCommands; private final Map, ServiceLoader> servicesLoaded = new ConcurrentHashMap<>(); + private final Set lifeCycleEventTypes; private WorkflowApplication(Builder builder) { this.taskFactory = builder.taskFactory; @@ -152,6 +155,7 @@ private WorkflowApplication(Builder builder) { this.lifeCycleCloudEventFactory = builder.lifeCycleCloudEventFactory; this.schedulerExecutorService = builder.schedulerExecutorService; this.allowedCommands = Collections.unmodifiableSet(builder.allowedCommands); + this.lifeCycleEventTypes = builder.lifeCycleEventTypes; } public TaskExecutorFactory taskFactory() { @@ -280,6 +284,7 @@ public SchemaValidator getValidator(SchemaInline inline) { private CronResolverFactory cronResolverFactory; private ScheduledExecutorService schedulerExecutorService; private Set allowedCommands = new HashSet<>(); + private Set lifeCycleEventTypes; private Builder() { ServiceLoader.load(NamedWorkflowAdditionalObject.class) @@ -458,6 +463,11 @@ public Builder withCronResolverFactory(CronResolverFactory cronResolverFactory) return this; } + public Builder withLifeCycleEventTypes(Set lifeCycleEventTypes) { + this.lifeCycleEventTypes = lifeCycleEventTypes; + return this; + } + public WorkflowApplication build() { if (executorFactory == null) { executorFactory = new DefaultExecutorServiceFactory(); @@ -565,7 +575,9 @@ public CronResolver parseCron(String cron) { if (authProviderFactory == null) { authProviderFactory = DefaultAuthProviderFactory.factory(); } - + if (lifeCycleEventTypes == null) { + lifeCycleEventTypes = EnumSet.of(LifeCycleEventType.V8); + } return new WorkflowApplication(this); } @@ -710,6 +722,10 @@ public Set allowedCommands() { return allowedCommands; } + public Set lifeCycleEventTypes() { + return lifeCycleEventTypes; + } + @SuppressWarnings("unchecked") public > List serviceLoadedClasses(Class clazz) { ServiceLoader serviceLoader = servicesLoaded.computeIfAbsent(clazz, ServiceLoader::load); diff --git a/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/EventType.java b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/EventType.java index ce64985ee..0ef4ec470 100644 --- a/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/EventType.java +++ b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/EventType.java @@ -15,29 +15,31 @@ */ package org.openworkflow.sdk.impl.lifecycle; -import org.openworkflow.sdk.impl.LifecycleEvents; - public enum EventType { - TASK_STARTED(LifecycleEvents.TASK_STARTED), - TASK_COMPLETED(LifecycleEvents.TASK_COMPLETED), - TASK_SUSPENDED(LifecycleEvents.TASK_SUSPENDED), - TASK_RESUMED(LifecycleEvents.TASK_RESUMED), - TASK_FAULTED(LifecycleEvents.TASK_FAULTED), - TASK_CANCELLED(LifecycleEvents.TASK_CANCELLED), - TASK_RETRIED(LifecycleEvents.TASK_RETRIED), - WORKFLOW_STARTED(LifecycleEvents.WORKFLOW_STARTED), - WORKFLOW_COMPLETED(LifecycleEvents.WORKFLOW_COMPLETED), - WORKFLOW_SUSPENDED(LifecycleEvents.WORKFLOW_SUSPENDED), - WORKFLOW_RESUMED(LifecycleEvents.WORKFLOW_RESUMED), - WORKFLOW_FAULTED(LifecycleEvents.WORKFLOW_FAULTED), - WORKFLOW_CANCELLED(LifecycleEvents.WORKFLOW_CANCELLED), - WORKFLOW_STATUS_CHANGED(LifecycleEvents.WORKFLOW_STATUS_CHANGED); + TASK_STARTED("task.started.v1"), + TASK_COMPLETED("task.completed.v1"), + TASK_SUSPENDED("task.suspended.v1"), + TASK_RESUMED("task.resumed.v1"), + TASK_FAULTED("task.faulted.v1"), + TASK_CANCELLED("task.cancelled.v1"), + TASK_RETRIED("task.retried.v1"), + WORKFLOW_STARTED("workflow.started.v1"), + WORKFLOW_COMPLETED("workflow.completed.v1"), + WORKFLOW_SUSPENDED("workflow.suspended.v1"), + WORKFLOW_RESUMED("workflow.resumed.v1"), + WORKFLOW_FAULTED("workflow.faulted.v1"), + WORKFLOW_CANCELLED("workflow.cancelled.v1"), + WORKFLOW_STATUS_CHANGED("workflow.status-changed.v1"); private final String lifecycleId; + public String type() { + return this.lifecycleId; + } + @Override public String toString() { - return this.lifecycleId; + return LifeCycleEventType.V8.fullEventType(this.lifecycleId); } EventType(String lifecycleId) { diff --git a/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/LifeCycleEventType.java b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/LifeCycleEventType.java new file mode 100644 index 000000000..2ed3f843e --- /dev/null +++ b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/LifeCycleEventType.java @@ -0,0 +1,35 @@ +/* + * Copyright 2020-Present The Open 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 org.openworkflow.sdk.impl.lifecycle; + +public enum LifeCycleEventType { + V8("org.open-workflow-specification."), + V7("io.serverlessworkflow."); + + private String prefix; + + public String prefix() { + return prefix; + } + + public String fullEventType(String eventType) { + return prefix + eventType; + } + + private LifeCycleEventType(String prefix) { + this.prefix = prefix; + } +} diff --git a/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/ce/AbstractLifeCyclePublisher.java b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/ce/AbstractLifeCyclePublisher.java index ffe1e6ad0..02a87a060 100644 --- a/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/ce/AbstractLifeCyclePublisher.java +++ b/impl/core/src/main/java/org/openworkflow/sdk/impl/lifecycle/ce/AbstractLifeCyclePublisher.java @@ -15,22 +15,6 @@ */ package org.openworkflow.sdk.impl.lifecycle.ce; -import static org.openworkflow.sdk.impl.LifecycleEvents.APPLICATION_ID_EXTENSION; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_CANCELLED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_COMPLETED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_FAULTED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_RESUMED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_RETRIED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_STARTED; -import static org.openworkflow.sdk.impl.LifecycleEvents.TASK_SUSPENDED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_CANCELLED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_COMPLETED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_FAULTED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_RESUMED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_STARTED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_STATUS_CHANGED; -import static org.openworkflow.sdk.impl.LifecycleEvents.WORKFLOW_SUSPENDED; - import io.cloudevents.CloudEvent; import io.cloudevents.CloudEventData; import io.cloudevents.core.builder.CloudEventBuilder; @@ -38,10 +22,13 @@ import io.cloudevents.core.data.PojoCloudEventData.ToBytes; import java.time.OffsetDateTime; import java.util.Collection; -import java.util.Set; -import java.util.function.Function; +import java.util.function.BiFunction; +import java.util.stream.Stream; +import org.openworkflow.sdk.impl.LifecycleEvents; import org.openworkflow.sdk.impl.WorkflowApplication; import org.openworkflow.sdk.impl.events.CloudEventUtils; +import org.openworkflow.sdk.impl.lifecycle.EventType; +import org.openworkflow.sdk.impl.lifecycle.LifeCycleEventType; import org.openworkflow.sdk.impl.lifecycle.TaskCancelledEvent; import org.openworkflow.sdk.impl.lifecycle.TaskCompletedEvent; import org.openworkflow.sdk.impl.lifecycle.TaskFailedEvent; @@ -62,21 +49,7 @@ public abstract class AbstractLifeCyclePublisher implements WorkflowExecutionListener { public static Collection getLifeCycleTypes() { - return Set.of( - TASK_STARTED, - TASK_COMPLETED, - TASK_SUSPENDED, - TASK_RESUMED, - TASK_FAULTED, - TASK_CANCELLED, - TASK_RETRIED, - WORKFLOW_STARTED, - WORKFLOW_COMPLETED, - WORKFLOW_SUSPENDED, - WORKFLOW_RESUMED, - WORKFLOW_FAULTED, - WORKFLOW_CANCELLED, - WORKFLOW_STATUS_CHANGED); + return Stream.of(EventType.values()).map(EventType::toString).toList(); } private WorkflowLifeCycleCloudEventFactory lifeCycleFactory(WorkflowEvent ev) { @@ -88,11 +61,11 @@ public void onTaskStarted(TaskStartedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_STARTED))); + .withType(type.fullEventType(EventType.TASK_STARTED.type())))); } @Override @@ -100,11 +73,11 @@ public void onTaskRetried(TaskRetriedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_RETRIED))); + .withType(type.fullEventType(EventType.TASK_RETRIED.type())))); } @Override @@ -112,11 +85,11 @@ public void onTaskCompleted(TaskCompletedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_COMPLETED))); + .withType(type.fullEventType(EventType.TASK_COMPLETED.type())))); } @Override @@ -124,11 +97,11 @@ public void onTaskSuspended(TaskSuspendedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_SUSPENDED))); + .withType(type.fullEventType(EventType.TASK_SUSPENDED.type())))); } @Override @@ -136,11 +109,11 @@ public void onTaskResumed(TaskResumedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_RESUMED))); + .withType(type.fullEventType(EventType.TASK_RESUMED.type())))); } @Override @@ -148,11 +121,11 @@ public void onTaskCancelled(TaskCancelledEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_CANCELLED))); + .withType(type.fullEventType(EventType.TASK_CANCELLED.type())))); } @Override @@ -160,11 +133,11 @@ public void onTaskFailed(TaskFailedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(TASK_FAULTED))); + .withType(type.fullEventType(EventType.TASK_FAULTED.type())))); } @Override @@ -172,11 +145,11 @@ public void onWorkflowStarted(WorkflowStartedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_STARTED))); + .withType(type.fullEventType(EventType.WORKFLOW_STARTED.type())))); } @Override @@ -184,11 +157,11 @@ public void onWorkflowSuspended(WorkflowSuspendedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_SUSPENDED))); + .withType(type.fullEventType(EventType.WORKFLOW_SUSPENDED.type())))); } @Override @@ -196,11 +169,11 @@ public void onWorkflowCancelled(WorkflowCancelledEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_CANCELLED))); + .withType(type.fullEventType(EventType.WORKFLOW_CANCELLED.type())))); } @Override @@ -208,11 +181,11 @@ public void onWorkflowResumed(WorkflowResumedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_RESUMED))); + .withType(type.fullEventType(EventType.WORKFLOW_RESUMED.type())))); } @Override @@ -220,11 +193,11 @@ public void onWorkflowCompleted(WorkflowCompletedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_COMPLETED))); + .withType(type.fullEventType(EventType.WORKFLOW_COMPLETED.type())))); } @Override @@ -232,11 +205,11 @@ public void onWorkflowFailed(WorkflowFailedEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_FAULTED))); + .withType(type.fullEventType(EventType.WORKFLOW_FAULTED.type())))); } @Override @@ -245,11 +218,11 @@ public void onWorkflowStatusChanged(WorkflowStatusEvent event) { WorkflowLifeCycleCloudEventFactory factory = lifeCycleFactory(event); publish( event, - ev -> + (ev, type) -> factory.build( builder(event) .withData(cloudEventData(factory.build(event), this::convert)) - .withType(WORKFLOW_STATUS_CHANGED))); + .withType(type.fullEventType(EventType.WORKFLOW_STATUS_CHANGED.type())))); } } @@ -311,10 +284,12 @@ protected byte[] convert(WorkflowStatusCEDataEvent data) { protected abstract byte[] convertToBytes(T data); - protected void publish(T ev, Function ceFunction) { + protected void publish( + T ev, BiFunction ceFunction) { WorkflowApplication appl = appl(ev); if (appl.isLifeCycleCEPublishingEnabled()) { - publish(appl, ceFunction.apply(ev)); + appl.lifeCycleEventTypes() + .forEach(eventType -> publish(appl, ceFunction.apply(ev, eventType))); } } @@ -334,7 +309,7 @@ private static CloudEventBuilder builder(WorkflowEvent ev) { .withId(CloudEventUtils.id()) .withSource(CloudEventUtils.source()) .withTime(OffsetDateTime.now()) - .withExtension(APPLICATION_ID_EXTENSION, appl(ev).id()); + .withExtension(LifecycleEvents.APPLICATION_ID_EXTENSION, appl(ev).id()); } private static WorkflowApplication appl(WorkflowEvent ev) {