Triggers, scheduling, and enterprise integration patterns

The course has so far started every workflow by hand. This final lesson covers the other ways a workflow starts, the three task types the static course did not exercise directly (REST_CONNECTOR, HAYSTACK, MQTT_PUBLISH), and a reference mapping of Twinit tasks onto standard enterprise integration patterns.

Triggers

A workflow definition is inert until something starts it. There are three triggers.

Manual Run

A direct API call: from an app, a page-component action, or a workbench. This is what runWeatherWorkflow and what the 'Run Workflow' command does throughout the course.

POST /workflowsvc/api/v1/workflows
{
    "_workflowDefId": "…",
    "_inputParams": { "_urn": "urn:…" }
}

Scheduled

Attach one or more Schedules to a WorkflowDef. Schedules are sub-resources managed through their own REST endpoints, using a cron expression:

{
    "_schedule": {
        "_type": "TIME_BASED",
        "_runInterval": "0 0/5 * ? * *",
        "_timeZone": "UTC"
    }
}

The example above runs the workflow every five minutes. This is how you turn any of the course workflows into an unattended job (e.g. poll a weather API every 15 minutes and store readings).

You can create Workflow Schedule using a user interface int he IDE Extension.

Expand the Workflow Definition in the Workflow Service to show the Schedules
Right click the Schedules node and select 'Create Schedule'

wf sched node

A web form will open with controls to make screaitng a schedule easy

wf sched ui

Event-Driven

Configure a Notification Service trigger that fires a custom script when a matching event is emitted anywhere in Twinit; the script calls IafWorkflowSvc.runWorkflow.

filter: { event: "resource.ResourceCreated", "resource.type": "filesvc.*" }
    → script → runWorkflow(defId, …)

This is how "when a file is uploaded, process it" style automation is built, and it is the same mechanism that would deliver the USER_INPUT pause notification to an approver.

The Three External Task Types

These task types require live endpoints, so the course did not cover them, but their shape is worth knowing because they are how workflows reach outside Twinit.

REST_CONNECTOR

The production counterpart of the course's generate...WeatherData scripts. It supports NoAuth, Basic, Bearer, and OAuth2, client-credentials, so most enterprise APIs need no custom auth code. Its result is exposed as ._output.result (or .body) for downstream binding.

{
    _name: "fetch",
    _type: "REST_CONNECTOR",
    _inputParams: {
        _url: "${workflow.input._url}",
        _auth: { _type: "OAuth2ClientCreds", /* ... */ },
        _args: { method: "GET", headers: { Accept: "application/json" } },
        to: "json"
    }
}

To convert any course workflow to live data, replace a generate...WeatherData SCRIPT_EXECUTION task with a REST_CONNECTOR task and update the downstream binding from ._output.scriptOutput to ._output.result. Everything else is unchanged.

It's also worth noting that fetch() can be used directly in a SCRIPT_EXECUTION task as well. Depending on your use case, fetch() may be a more flexible implementation than the REST_CONNECTOR task.

HAYSTACK

A first-class task for SkySpark, Niagara, and other Haystack-compliant BMS endpoints. It supports actions such as pointhistory:

{
    _name: "fetch_ahu_history", 
    _type: "HAYSTACK",
    _sequenceno: 1,
    _inputParams: {
        _project: "site-A",
        _url: "https://skyspark.example.com/api",
        _uname: "twinit", 
        _pwd: "${secret:skyspark_pwd}",
        _action: "pointhistory",
        _cmd: "point and his and equip->ahu",
        _hisStartDate: "${workflow.input._since}",
        _to: "json"
    }
}

Combine it with FORK_JOIN to fetch history for several point groups at once, then a SCRIPT_EXECUTION to normalise each history frame into telemetry readings (_ts + _tsMetadata._sourceId).

MQTT_PUBLISH

Publishes a message to any MQTT broker (ActiveMQ, HiveMQ, AWS IoT Core, or a BACnet-MQTT gateway). Used for command messages to devices:

{
    _name: "publish_setpoint", 
    _type: "MQTT_PUBLISH",
    _inputParams: {
        _mqtt_target: [ { _config: { _url: "ssl://gw.example.com:8883",
                                     _username: "twinit-bms",
                                     _password: "${secret:bldg_a_pwd}",
                                     _mqtt_version: 5 } } ],
        _topic: "bldg/A/ahu/03/setpoint",
        _messages: [ "{ \"objectId\": \"analog-value:12\", \"value\": ${workflow.input.setpointC}, \"correlationId\": \"${workflow._id}\" }" ]
    }
}

Note correlationId: "${workflow._id}": the run id correlates the command with the device's echoed state on a separate topic, which flows back through Twinit's IoT ingest pipeline.

Task-to-Pattern Reference

Twinit's built-in tasks map directly onto the most-used enterprise integration patterns. This table is the quick reference for choosing a task:

Pattern	Twinit implementation	Course lesson
Router	`SWITCH` on `${prev._output.field}`	06
Recipient List	`FORK_JOIN` of `REST_CONNECTOR`s	05
Splitter + Aggregator	`SCRIPT_EXECUTION` (fan-out) → `FORK_JOIN` (join)	07
Filter	`SWITCH` with a drop branch (or in-script decision)	06, 08
Message Translator	`SCRIPT_EXECUTION` (map schema A → B)	09
Enricher	`REST_CONNECTOR` → `SCRIPT_EXECUTION` (merge)	07
Poll	`DO_WHILE` around `REST_CONNECTOR`	08
Command Message	`MQTT_PUBLISH` with `correlationId`	13
Request–Reply / Human Task	`REST_CONNECTOR` + `USER_INPUT`	10
Process Manager	the workflow itself (25-task envelope)	all
AI-backed Translator / Router	`AI_CONVERSE`	11

Triggers​

Manual Run​

Scheduled​

Event-Driven​

The Three External Task Types​

REST_CONNECTOR​

HAYSTACK​

MQTT_PUBLISH​

Task-to-Pattern Reference​