1"use strict";(self.webpackChunkwindmilldocs=self.webpackChunkwindmilldocs||[]).push([["10929"],{90065(e,r,n){n.r(r),n.d(r,{metadata:()=>t,default:()=>h,frontMatter:()=>o,contentTitle:()=>a,toc:()=>c,assets:()=>l});var t=JSON.parse('{"id":"triggers/mqtt_triggers/index","title":"MQTT triggers","description":"How do I trigger scripts and flows from MQTT broker messages in Windmill?","source":"@site/docs/triggers/8_mqtt_triggers/index.mdx","sourceDirName":"triggers/8_mqtt_triggers","slug":"/triggers/mqtt_triggers/","permalink":"/docs/triggers/mqtt_triggers/","draft":false,"unlisted":false,"tags":[],"version":"current","frontMatter":{"description":"How do I trigger scripts and flows from MQTT broker messages in Windmill?"},"sidebar":"tutorialSidebar","previous":{"title":"SQS","permalink":"/docs/triggers/sqs_triggers/"},"next":{"title":"GCP","permalink":"/docs/triggers/gcp_triggers/"}}'),s=n(74848),i=n(28453);let o={description:"How do I trigger scripts and flows from MQTT broker messages in Windmill?"},a="MQTT triggers",l={},c=[{value:"How to use",id:"how-to-use",level:2},{value:"Configure MQTT resource",id:"configure-mqtt-resource",level:3},{value:"Select runnable",id:"select-runnable",level:3},{value:"Configure topic subscriptions",id:"configure-topic-subscriptions",level:3},{value:"Quality of Service (QoS) levels",id:"quality-of-service-qos-levels",level:4},{value:"MQTT topic structure",id:"mqtt-topic-structure",level:4},{value:"Advanced MQTT options",id:"advanced-mqtt-options",level:3},{value:"Implementation examples",id:"implementation-examples",level:2},{value:"Basic script",id:"basic-script",level:3},{value:"Script with preprocessor",id:"script-with-preprocessor",level:3},{value:"MQTT object",id:"mqtt-object",level:4},{value:"MQTT v5 properties",id:"mqtt-v5-properties",level:4},{value:"Error handling",id:"error-handling",level:2}];function d(e){let r={a:"a",br:"br",code:"code",h1:"h1",h2:"h2",h3:"h3",h4:"h4",header:"header",li:"li",p:"p",pre:"pre",strong:"strong",table:"table",tbody:"tbody",td:"td",th:"th",thead:"thead",tr:"tr",ul:"ul",...(0,i.R)(),...e.components};return(0,s.jsxs)(s.Fragment,{children:[(0,s.jsx)(r.header,{children:(0,s.jsx)(r.h1,{id:"mqtt-triggers",children:"MQTT triggers"})}),"\n",(0,s.jsxs)(r.p,{children:["Windmill can connect to an ",(0,s.jsx)(r.a,{href:"https://mqtt.org/mqtt-specification/",children:(0,s.jsx)(r.strong,{children:"MQTT"})})," broker and trigger runnables (scripts, flows) in response to messages published to specified topics. MQTT triggers are not available on the ",(0,s.jsx)(r.a,{href:"/pricing",children:"Cloud"}),"."]}),"\n",(0,s.jsx)("iframe",{style:{aspectRatio:"16/9"},src:"https://www.youtube.com/embed/T_ava06lqlc?vq=1080p",title:"MQTT triggers",frameBorder:"0",allow:"accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share",allowFullScreen:!0,className:"border-2 rounded-lg object-cover w-full dark:border-gray-800"}),"\n",(0,s.jsx)(r.h2,{id:"how-to-use",children:"How to use"}),"\n",(0,s.jsx)(r.h3,{id:"configure-mqtt-resource",children:"Configure MQTT resource"}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:["Select an existing ",(0,s.jsx)(r.a,{href:"https://hub.windmill.dev/resource_types/225/mqtt",children:"MQTT resource"})," or create a new one"]}),"\n",(0,s.jsx)(r.li,{children:"Provide broker hostname and port"}),"\n",(0,s.jsx)(r.li,{children:"Add authentication credentials and certificates as required by your broker"}),"\n"]}),"\n",(0,s.jsx)(r.h3,{id:"select-runnable",children:"Select runnable"}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsx)(r.li,{children:"Choose the script or flow to execute when messages are published to your subscribed topics"}),"\n"]}),"\n",(0,s.jsx)(r.h3,{id:"configure-topic-subscriptions",children:"Configure topic subscriptions"}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsx)(r.li,{children:"Specify one or more topics to subscribe to"}),"\n",(0,s.jsx)(r.li,{children:"Set appropriate QoS level for each topic"}),"\n"]}),"\n",(0,s.jsx)(r.h4,{id:"quality-of-service-qos-levels",children:"Quality of Service (QoS) levels"}),"\n",(0,s.jsxs)(r.table,{children:[(0,s.jsx)(r.thead,{children:(0,s.jsxs)(r.tr,{children:[(0,s.jsx)(r.th,{children:"Level"}),(0,s.jsx)(r.th,{children:"Description"}),(0,s.jsx)(r.th,{children:"When to use"})]})}),(0,s.jsxs)(r.tbody,{children:[(0,s.jsxs)(r.tr,{children:[(0,s.jsx)(r.td,{children:(0,s.jsx)(r.strong,{children:"0"})}),(0,s.jsxs)(r.td,{children:[(0,s.jsx)(r.strong,{children:"At most once"})," \u2013 Message delivered once or not at all without confirmation"]}),(0,s.jsx)(r.td,{children:"Choose when it is okay for your script/flow to not be triggered (if the message is lost) or triggered only once."})]}),(0,s.jsxs)(r.tr,{children:[(0,s.jsx)(r.td,{children:(0,s.jsx)(r.strong,{children:"1"})}),(0,s.jsxs)(r.td,{children:[(0,s.jsx)(r.strong,{children:"At least once"})," \u2013 Guaranteed delivery but may arrive multiple times"]}),(0,s.jsx)(r.td,{children:"Choose when it is okay for your script/flow to be triggered again by an already received message from the broker."})]}),(0,s.jsxs)(r.tr,{children:[(0,s.jsx)(r.td,{children:(0,s.jsx)(r.strong,{children:"2"})}),(0,s.jsxs)(r.td,{children:[(0,s.jsx)(r.strong,{children:"Exactly once"})," \u2013 Guaranteed delivery exactly once"]}),(0,s.jsx)(r.td,{children:"Choose when you need your script/flow to be triggered only once and avoid any duplicates."})]})]})]}),"\n",(0,s.jsxs)(r.p,{children:["For more information about MQTT QoS, see the ",(0,s.jsx)(r.a,{href:"https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels/",children:"MQTT QoS Documentation"}),"."]}),"\n",(0,s.jsx)(r.h4,{id:"mqtt-topic-structure",children:"MQTT topic structure"}),"\n",(0,s.jsxs)(r.p,{children:["MQTT topics are case-sensitive and follow a hierarchical structure (e.g., ",(0,s.jsx)(r.code,{children:"home/sensor/temperature"}),").",(0,s.jsx)(r.br,{}),"\n","For best practices on MQTT topics, see the ",(0,s.jsx)(r.a,{href:"https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/",children:"MQTT Topics Documentation"}),"."]}),"\n",(0,s.jsx)(r.h3,{id:"advanced-mqtt-options",children:"Advanced MQTT options"}),"\n",(0,s.jsxs)(r.p,{children:["By default, Windmill uses ",(0,s.jsx)(r.strong,{children:"MQTT version 5"}),". However, you can choose to use ",(0,s.jsx)(r.strong,{children:"MQTT version 3"})," or ",(0,s.jsx)(r.strong,{children:"MQTT version 5"})," with specific associated options."]}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"MQTT v3 options"}),":","\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Clean Session"})," (default: true): ",(0,s.jsx)(r.a,{href:"https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#clean-session-in-mqtt-3-1-1",children:"Learn more"})]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Client ID"}),": ",(0,s.jsx)(r.a,{href:"https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html",children:"Learn more"})]}),"\n"]}),"\n"]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"MQTT v5 options"}),":","\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Clean Start"})," (default: true): ",(0,s.jsx)(r.a,{href:"https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#introduction-to-clean-start",children:"Learn more"})]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Session Expiry Interval"}),": ",(0,s.jsx)(r.a,{href:"https://www.emqx.com/en/blog/mqtt5-new-feature-clean-start-and-session-expiry-interval#introduction-to-session-expiry-interval",children:"Learn more"})]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Topic Alias Maximum"}),": ",(0,s.jsx)(r.a,{href:"https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901051",children:"Learn more"})]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:"Client ID"}),": ",(0,s.jsx)(r.a,{href:"https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901059",children:"Learn more"})]}),"\n"]}),"\n"]}),"\n"]}),"\n",(0,s.jsx)(r.h2,{id:"implementation-examples",children:"Implementation examples"}),"\n",(0,s.jsx)(r.p,{children:"Below are code examples demonstrating how to handle MQTT messages in your Windmill scripts. You can either process messages directly in a basic script or use a preprocessor for more advanced message handling and transformation before execution."}),"\n",(0,s.jsx)(r.h3,{id:"basic-script",children:"Basic script"}),"\n",(0,s.jsx)(r.pre,{children:(0,s.jsx)(r.code,{className:"language-typescript",children:'export async function main(payload: string) {\n // Convert base64 encoded payload to string\n const textPayload = atob(payload);\n \n // Parse JSON if applicable\n try {\n const jsonData = JSON.parse(textPayload);\n console.log("Received JSON data:", jsonData);\n // Process JSON data\n } catch (e) {\n // Handle as plain text\n console.log("Received text data:", textPayload);\n // Process text data\n }\n \n return { processed: true, message: "MQTT message processed successfully" };\n}\n'})}),"\n",(0,s.jsx)(r.h3,{id:"script-with-preprocessor",children:"Script with preprocessor"}),"\n",(0,s.jsxs)(r.p,{children:["If you use a ",(0,s.jsx)(r.a,{href:"/docs/core_concepts/preprocessors/",children:"preprocessor"}),", the preprocessor function receives the message payload as base64 encoded string and an MQTT object with the following fields:"]}),"\n",(0,s.jsx)(r.h4,{id:"mqtt-object",children:"MQTT object"}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"topic"})}),": The MQTT topic on which the message was received."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"retain"})}),": Boolean indicating if the message is retained."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"pkid"})}),": Packet identifier (if QoS > 0)."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"qos"})}),": Quality of Service level."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"v5"})}),": MQTT v5 properties (optional)."]}),"\n"]}),"\n",(0,s.jsx)(r.h4,{id:"mqtt-v5-properties",children:"MQTT v5 properties"}),"\n",(0,s.jsxs)(r.ul,{children:["\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"payload_format_indicator"})}),": Indicates if the payload is UTF-8 encoded or binary."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"topic_alias"})}),": An alias for the topic name."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"response_topic"})}),": A topic for the recipient to send a response to."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"correlation_data"})}),": Correlation data for request/response."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"user_properties"})}),": A list of user-defined properties."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"subscription_identifiers"})}),": Subscription identifiers."]}),"\n",(0,s.jsxs)(r.li,{children:[(0,s.jsx)(r.strong,{children:(0,s.jsx)(r.code,{children:"content_type"})}),": The content type of the payload."]}),"\n"]}),"\n",(0,s.jsxs)(r.p,{children:["For more information about MQTT v5 properties, see the ",(0,s.jsx)(r.a,{href:"https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901109",children:"MQTT v5 Properties Documentation"}),"."]}),"\n",(0,s.jsx)(r.pre,{children:(0,s.jsx)(r.code,{className:"language-typescript",children:"/**\n * General Trigger Preprocessor\n * \n * \u26A0\uFE0F This function runs BEFORE the main function.\n * \n * It processes raw trigger data (e.g., MQTT, HTTP, SQS) before passing it to `main()`.\n * Common tasks:\n * - Convert binary payloads to string/JSON\n * - Extract metadata\n * - Filter messages\n * - Add timestamps/context\n * \n * The returned object determines `main()` parameters:\n * - `{a: 1, b: 2}` \u2192 `main(a, b)`\n * - `{payload}
1` \u2192 `main(payload)`\n * \n * @param event - Trigger data and metadata (e.g., MQTT, HTTP)\n * @returns Processed data for `main()`\n */\nexport async function preprocessor(\n event: {\n kind: \"mqtt\",\n payload: string, // base64 encoded payload\n topic: string,\n retain: boolean,\n pkid: number,\n qos: number,\n v5?: {\n payload_format_indicator?: number,\n topic_alias?: number,\n response_topic?: string,\n correlation_data?: Array<number>,\n user_properties?: Array<[string, string]>,\n subscription_identifiers?: Array<number>,\n content_type?: string\n }\n\n }\n) {\n if (event.kind === 'mqtt') {\n const payloadAsString = atob(event.payload);\n const uint8Payload = Uint8Array.from(payloadAsString, (c) => c.charCodeAt(0));\n\n return {\n contentType: event.v5?.content_type,\n payload: uint8Payload,\n payloadAsString\n };\n }\n // We assume the script is triggered by an MQTT message, which is why an error is thrown for other trigger kinds.\n // If the script is intended to support other triggers, update this logic to handle the respective trigger kind.\n throw new Error(`Expected mqtt trigger kind got: ${event.kind}`)\n}\n\n/**\n * Main Function - Handles processed trigger events\n * \n * \u26A0\uFE0F Called AFTER `preprocessor()`, with its return values.\n * \n * @param payload - Raw binary payload\n * @param payloadAsString - Decoded string payload\n * @param contentType - MQTT v5 content type (if available)\n */\nexport async function main(payload: Uint8Array, payloadAsString: string, contentType?: string) {\n\n}\n"})}),"\n",(0,s.jsx)(r.h2,{id:"error-handling",children:"Error handling"}),"\n",(0,s.jsxs)(r.p,{children:["MQTT triggers support local error handlers that override workspace error handlers for specific triggers. See the ",(0,s.jsx)(r.a,{href:"/docs/core_concepts/error_handling/#trigger-error-handlers",children:"error handling documentation"})," for configuration details and examples."]})]})}function h(e={}){let{wrapper:r}={...(0,i.R)(),...e.components};return r?(0,s.jsx)(r,{...e,children:(0,s.jsx)(d,{...e})}):d(e)}},28453(e,r,n){n.d(r,{R:()=>o,x:()=>a});var t=n(96540);let s={},i=t.createContext(s);function o(e){let r=t.useContext(i);return t.useMemo(function(){return"function"==typeof e?e(r):{...r,...e}},[r,e])}function a(e){let r;return r=e.disableParentContext?"function"==typeof e.components?e.components(s):e.components||s:o(e.components),t.createElement(i.Provider,{value:r},e.children)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.