Trigger Connector
The Trigger Connector allows you to start a Business Process workflow based on external events or scheduled polling of an external source system, like an incoming email, a new message in a queue, a new file for processing in a folder, and so on. To achieve this, apply:
The Trigger element in the Business Process designer. Its configuration is similar to the Java native bot step but contains a different config type:
trigger. The configuration XML structure is as follows:<?xml version="1.0" encoding="UTF-8"?>
<config type="trigger">
<worker>com.test:trigger-test:1.0</worker>
<processor>sample-processor</processor>
<configuration>10000</configuration>
</config>Where:
The
<worker>tag is your Trigger Worker's actualgroupId:artifactId:version(see in the artifact'spom.xmlfile).The
<processor>tag is the corresponding pipeline processor ID. See theidattribute in the@PipelineProcessorannotation on the corresponding class.The
<configuration>tag is theStringpipeline configuration passed as a part ofPipelineContexton the processor's start.
The Trigger Connector Worker type. For more details, see the sections below.
Develop trigger
To create a Trigger Connector project, perform the following steps:
Generate a Trigger Worker app from the archetype.
Add needed modules and third-party dependencies.
To apply the trigger pipeline processor functionality, implement the
IPipelineProcessorinterface and add the required logic as Java classes. Then, configure unit tests to cover the functionality.Create, verify, and change the trigger bot configuration inside the artifact.
Generate trigger app from archetype
Before you generate a Trigger Worker application, you should prepare the environment. For instructions, see Prepare environment for archetype generation.
A generation sample of the Trigger Worker project is similar to the Java Native Worker. A special Maven trigger-worker-archetype project was created for building a new Trigger Worker app.
- In the parent folder for the target Java Native Worker app, run the following command. It can be the root folder of your AI Agent project or an empty folder if you want to develop the Worker as a standalone project. Replace the
${version}placeholder with the current trigger-worker-archetype project version.
mvn archetype:generate -DarchetypeGroupId=com.workfusion.spa.core -DarchetypeArtifactId=trigger-worker-archetype -DarchetypeVersion=${version}
Specify the following properties for the Maven archetype generation mechanism to generate a new project:
groupId: trigger's Maven group IDartifactId: trigger's Maven artifact ID; also the folder and target artifact name by defaultversion: trigger's Maven versionpackage: root package for trigger Java classes
For details on how to use archetype, see the
readme.mdfile in the trigger-worker-archetype project.After you save the properties, a new project is generated in the subfolder. The subfolder's name should correspond to
artifactId. The generated app is a Spring Boot application with minimum dependencies required to implement the base Trigger Worker logic.
Add modules and third-party dependencies
By default, no modules are added to the dependency in the generated app. Only one out-of-the-box module-secrets module is available, which is used to access secure entries from Secrets Vault. It is the same module as for the Java Native Worker.
You can access configuration parameters from Secrets Vault by default using standard Spring mechanisms, for example, the @Value annotation. There is no need to access this module unless you must access custom secure entries.
The example contains the dependency added to pom.xml but commented:
<dependencies>
<!--uncomment the following dependency if a module that works with client secure entries is needed-->
<!--<dependency>
<groupId>com.workfusion.spa.java.native.worker</groupId>
<artifactId>module-secrets</artifactId>
<version>${wf.module-secrets.version}</version>
<exclusions>
<exclusion>
<groupId>com.workfusion.spa.java.native.worker</groupId>
<artifactId>java-native-worker-core</artifactId>
</exclusion>
</exclusions>
</dependency>-->
</dependencies>
To use the module-secrets module, uncomment it.
Keep <exclusion> for java-native-worker-core as it is required to maintain compatibility between the Trigger Connector framework and the Java Native Worker module.
You can add required third-party dependencies manually in the same way as you add any other Maven dependency:
<dependencies>
<!-- ... -->
<dependency>
<groupId>com.someproject.group</groupId>
<artifactId>artifact-with-some-functionality</artifactId>
<version>1.2.3</version>
</dependency>
</dependencies>
Implement trigger pipeline processor functionality
The pipeline processor contains the main trigger logic. To apply the processor functionality, add IPipelineProcessor classes—one class per each pipeline type. The interface requires two methods to be implemented:
startPipeline(PipelineContext pipelineContext)retrieves all needed information from passedPipelineContext(for example, configuration, pre-configured task submitter, and so on) and starts pipeline processing. It is important that the processing logic is started asynchronously (as a separate thread) and does not block this method.stopPipeline()stops pipeline processing and performs cleanup operations if needed.
The implementation should support a start-after-stop sequence as the processor can be restarted if the pipeline configuration is changed.
To make a created processor class available in the context, annotate it with @PipelineProcessor(id = "sample-processor"), where id is the processor's name.
Each trigger can contain multiple processors if you need to provide a multi-functional artifact. IDs for processors across a single artifact should be unique. A trigger artifact containing more than one processor with the same ID is invalid and fails to start.
The processor is a Spring component that allows you to use any Spring mechanism, like auto-wiring, post-construct and per-destroy, access environment and configuration, and so on.
Send records
There are two types of record processing:
Asynchronous. The trigger sends a record and receives a confirmation response only if the record is accepted during the initial request.
A
RecordResultresponse object is returned withrequestIdand the current processing status. If the record sending fails, error status details are filled.Synchronous. The trigger sends a record and waits for its processing result.
A
RecordResultresponse object is returned and contains the result data in case of a successful execution or error status details if the record processing or sending fails.
The out-of-the-box TaskSubmitter is available in PipelineContext and has the following sending methods:
RecordResult sendRecord(Map<String, Object> data)sends a record to the asynchronous processing.RecordResult sendRecordSync(Map<String, Object> data)sends a record to the synchronous processing.RecordResult checkAsyncRecordStatus(String requestId)checks the asynchronous record processing status.void sendRecord(Map<String, Object> data, Consumer<RecordResult> resultCallback)sends a record to the asynchronous processing and checks the processing result in the background. A result callback should be provided and is performed once the processing result is ready.
Depending on your needs, you can use the asynchronous processing in two scenarios:
A record processing result is not required. Send a record and verify if it is accepted by Control Tower by checking
RecordProcessingStatusat the returnedRecordResultobject.A record processing result is required. After you send a record, you get
requestIdfromRecordResultand implement the logic to periodically check a record processing status by calling thecheckAsyncRecordStatusout-of-the-box method withrequestId. Once the record processing is finished, a check status call returnsRecordResultcontaining the result data in case of a successful execution or error status details if the processing fails.
If you need to send a record processing result to the trigger, your Business Process must have either a WebHarvest-based step with the send-to-external-connector="true" export plugin attribute or a Java Native Worker-based step with a sendResultToCaller Boolean parameter set to "true". The step output is sent back to the connector and processed based on the request type. For details, see the step config examples below.
A sample code for a WebHarvest-based step for sending results to the trigger:
<?xml version="1.0" encoding="UTF-8"?>
<config charset="UTF-8">
<export include-original-data="false" send-to-external-connector="true">
</export>
</config>
A sample code for a Java Native Worker-based step for sending results to the trigger:
<?xml version="1.0" encoding="UTF-8"?>
<config type="java">
<worker>my-company:worker:1.0.0</worker>
<processor>my-custom-task</processor>
<sendResultToCaller>true</sendResultToCaller>
</config>
In the generated app, the archetype creates SamplePipelineProcessor that you can use as an example of how to implement the pipeline processor:
@PipelineProcessor(id = "sample-processor")
public class SamplePipelineProcessor implements IPipelineProcessor {
private static final Logger logger = LoggerFactory.getLogger(SamplePipelineProcessor.class);
private ExecutorService executor;
private TaskSubmitter taskSubmitter;
private volatile boolean running;
@PreDestroy
public void tearDown() {
if (running) {
executor.shutdownNow();
}
}
@Override
public void startPipeline(PipelineContext pipelineContext) {
// Implement the logic for starting a pipeline.
// WARNING: do not execute the pipeline logic in this thread (do not block it); start it asynchronously!
// Example:
executor = Executors.newSingleThreadExecutor();
// parse config from string and read needed data
String config = pipelineContext.getPipeline().getPipelineConfig();
// other config string transformations if needed
// retrieve the pre-configured task submitter from context
taskSubmitter = pipelineContext.getTaskSubmitter();
running = true;
// asynchronously start pipeline processing
executor.submit(() -> processPipeline(config));
}
@Override
public void stopPipeline() {
// implement the pipeline stop and clean-up operations (close connections, and so on).
// NOTE: the trigger framework can call this method when it needs to either:
// 1) Stop the pipeline when its state changes in the configuration; then, you can restart the pipeline by calling the startPipeline method with the updated config.
// 2) Stop the pipeline before the Trigger Worker is destroyed.
// Example:
running = false;
try {
executor.shutdown();
if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
// some logic, if necessary, in case of pipeline processing cannot be terminated
Thread.currentThread().interrupt();
} finally {
// other clean up: close connections, and so on.
}
}
private void processPipeline(String config) {
while (running) {
try {
// Do the necessary logic of preparing or generating input data for the Business Process, for example:
Map<String, Object> data = new HashMap<>();
data.put("some", "data");
// Send input and receive a record-sending result from Control Tower. The method can throw TriggerRecordSendingException.
// NOTE: you will receive an async record call result representing whether it was accepted by Control Tower or not (status, details, and so on).
// It is important to process a record-sending result and check the status (should be "IN_PROGRESS") to verify if the record was accepted by Control Tower.
// It is a developer's responsibility to handle failures (a negative sending result status or an exception inside the pipeline processing thread) correctly if the data was taken (consumed)
// from some external system, and it cannot be read for one more time; the processing logic should return data, uncommit intake, or perform any other action to guarantee that the taken data will not be lost.
// The Jackson serializer or deserializer is used; it means your data must have a correct POJO object with a getter or setter that can be serialized by Jackson.
RecordResult recordResult = taskSubmitter.sendRecord(data);
/* record result status: RecordProcessingStatus status = */ recordResult.getStatus();
} catch (Exception e) {
logger.error(e.getMessage(), e);
// Process exceptions here and perform actions to prevent data loss if the intake is from some external system.
}
}
}
}
Modify trigger bot configuration in artifact
To import the trigger bot configuration directly with the trigger artifact, see the Import as part of AI Agent Asset Bundle section.
The bot configuration should exist in the form of an XML file in the src/main/resources/configs/main folder of your trigger project, for example:
my-trigger-10.2.9
└──src
└──main
└──resources
└──configs
└──main
├──sample-processor.xml
└──email-intake-processor.xml
When a trigger artifact is built and imported as part of the Asset Bundle, Bot Configs are also imported to Control Tower. In the trigger app generated from the archetype, see the sample-processor.xml bot configuration as an example.
Expose trigger for external system
To expose the trigger for external REST API calls, modify its connector.yml configuration file by adding the exposed: true parameter, for example:
command: ${env.parameters} ${java} -XX:CICompilerCount=2 -XX:+UseSerialGC -XX:MaxMetaspaceSize=${metaspace}M -Xmx${heap}M -XX:+ExitOnOutOfMemoryError -Dfile.encoding=UTF8 -Dlogging.file.path=${log.dir} -Dlogging.file.name=${log.file} -Djava.io.tmpdir=${temp.dir} -Dconnector.uuid=${connector.uuid} ${jvm.parameters} -jar ${connector.jar} ${config.server.parameters}
cpu: 0.4
memory:
heap: 512
metaspace: 128
health-checks:
- protocol: MESOS_HTTP
path: /actuator/health
grace-period-sec: 180
interval-sec: 60
timeout-sec: 10
max-failures: 3
exposed: true
You can expose the trigger with the base path constructed automatically. You can also set the static exposure base path for this artifact via the exposure-base-path property in connector.yml.
The system cannot have more than one exposed trigger with the same static base path. In case of a collision, the error mode is activated with a message for the misconfigured BP.
See the example below:
command: ${env.parameters} ${java} -XX:CICompilerCount=2 -XX:+UseSerialGC -XX:MaxMetaspaceSize=${metaspace}M -Xmx${heap}M -XX:+ExitOnOutOfMemoryError -Dfile.encoding=UTF8 -Dlogging.file.path=${log.dir} -Dlogging.file.name=${log.file} -Djava.io.tmpdir=${temp.dir} -Dconnector.uuid=${connector.uuid} ${jvm.parameters} -jar ${connector.jar} ${config.server.parameters}
cpu: 0.4
memory:
heap: 512
metaspace: 128
health-checks:
- protocol: MESOS_HTTP
path: /actuator/health
grace-period-sec: 180
interval-sec: 60
timeout-sec: 10
max-failures: 3
exposed: true
exposure-base-path: my-custom-path-here
You can override the static base path in the trigger artifact using the <launchConfiguration> tag for a trigger element in a BP. The value for this tag should be a JSON string, for example:
<?xml version="1.0" encoding="UTF-8"?>
<config type="trigger">
<worker>com.test:trigger-test:1.0</worker>
<processor>sample-processor</processor>
<configuration>10000</configuration>
<launchConfiguration>{"exposure-base-path":"my-override-custom-path"}</launchConfiguration>
</config>
Since the trigger app is exposed without the static exposure base path, you cannot control its base context path. It is calculated and set during the trigger deployment and contains a base prefix and a definition UUID.
The URL pattern for exposed REST API is https://${domain_dns}/connectors/definition_${GAV.replaceAll(“.“, "_").replaceAll(":", "_")}_${profile}_${BP definition}/${TRIGGER_REST_API_PATH}.
Example 1
If:
- Business Process definition:
82373b6e-3fd1-4bea-ba78-77325d48a416 - Trigger artifact GAV:
com.wf.demo:trigger-input:1.0.0 - REST API path:
input-data/generate/{data}
Then, the URL is as follows: https://${domain_dns}/connectors/definition_com_wf_demo_trigger-input_1_0_0_82373b6e-3fd1-4bea-ba78-77325d48a416/input-data/generate/{data}
Example 2
If:
- Business Process definition:
82373b6e-3fd1-4bea-ba78-77325d48a416 - Trigger artifact GAV:
com.wf.demo:trigger-input:1.0.0:profile-A - REST API path:
input-data/generate/{data}
Then, the URL is as follows: https://${domain_dns}/connectors/definition_com_wf_demo_trigger-input_1_0_0_profile-A_82373b6e-3fd1-4bea-ba78-77325d48a416/input-data/generate/{data}
If the static exposure base path is set, the URL pattern is as follows: https://${domain_dns}/connectors/${your-custom-path}/${TRIGGER_REST_API_PATH}
Implement the needed logic, for example, REST controllers, to provide endpoints to be accessed by an external system.
Authentication and authorization of exposed endpoints are the responsibilities of a Trigger Connector developer.
Build trigger artifact
To build a developed Trigger Worker artifact, use the standard Maven mvn clean install command. After a successful build, in the target folder, you get the Spring Boot application artifact as a JAR file that you can deploy to the target environment as any other Worker's artifacts.
Import trigger artifact
There are two options to deploy the trigger artifact to the Work.AI platform:
- Manually deploy to Nexus.
- Import as a part of an AI Agent Asset Bundle.
Manually deploy to Nexus
You can deploy the Worker's standard artifact to Nexus via the user interface or by running the Maven deploy command with custom parameters. In both cases, you must have credentials for a user with the deploy right to the connectors repository.
Import as part of AI Agent Asset Bundle
Before the actual import, assemble your Asset Bundle. For more details on the AI Agent Asset Bundle structure, see Package assets into AI Agent Asset Bundle.
To assemble an Asset Bundle with a trigger artifact, follow the steps below:
Create the folder structure under
artifactory-dependency/connectorsthat represents the GAV of the created trigger:artifactory-dependency
│ └── connectors
│ │ └── com
│ │ ├── workfusion
│ │ └── trigger-example
│ │ └── 1.0
│ │ └── trigger-example-1.0.jar
meta-info.jsonThe
meta-info.jsonexample is as follows:{
"AUTHOR": "Your Name",
"DESCRIPTION": "Trigger in bundle example",
"INSTANCE": "someinstance.workfusion.com",
"NAME": "trigger-example-1.0",
"PACKAGE_DATE": "2023-10-28 00:00:00",
"TARGET_VERSION": "10.2.8",
"WF_VERSION": "10.2.8"
}Create a zip archive with files and folders. Now, the Asset Bundle is assembled and ready to import.
To import, use the same import mechanism as for the Java Native Worker or BCB.
There is a difference from the Java Native Worker import procedure. The repository name for the Trigger Connector Worker should be connectors. It is a release repository, and you cannot import an artifact that contains the SNAPSHOT postfix in its version. Only release artifact versions are required.
After you import the trigger artifact, you can use it as the Trigger element when creating your Business Process. For details, see Start Business Process with no-code Trigger element.