> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/data-flows/service-commissioning-files/resources/cybus-endpoint.md). # Cybus::Endpoint Defining and managing data endpoints. A **Cybus::Endpoint** resource represents the address of a single data endpoint in the context of a specific protocol. Each endpoint is always mapped to exactly one internal MQTT topic. The MQTT topic can either be: * Explicitly defined using the [topic](#topic) property, or * Automatically generated (see note at [topic](#topic)). In most services, endpoint topics are further mapped via a [Cybus::Mapping](/data-flows/service-commissioning-files/resources/cybus-mapping.md) resource to application-specific topics, which can then be consumed by dashboards, applications, or other services. The actual protocol-specific data address is configured under the [subscribe, read, or write](#subscribe-read-write) properties. These properties are one level deeper in the YAML structure (beneath endpoint `subscribe`/`read`/`write`) and must not be confused with properties defined directly on the endpoint resource itself. ## Operation Results Endpoints of type `read` and `write` publish the results of their operations on an additional MQTT topic with the suffix `/res`: * A `read` endpoint named `myEndpoint` listens for requests on the MQTT topic `myEndpoint/req` and publishes the result on `myEndpoint/res`. * A `write` endpoint named `myEndpoint` listens for messages on the MQTT topic `myEndpoint/set` and publishes the result on `myEndpoint/res`. To match a result to your request, add an `id` to the request. Whether the result contains this `id` depends on the protocol, and SQL reads accept an `id` only in one case, see [Result Format per Protocol](#result-format-per-protocol). {% code lineNumbers="true" %} ```json { "id": "29194" } ``` {% endcode %} If the endpoint has [rules](#rules), Connectware applies them to the result before publishing it. ### Result Format per Protocol Each protocol defines the format of its results. The following table lists the properties of a successful result. A result contains `id` only where its cell lists it, and then only if your request contains one. | Protocol | `read` result | `write` result | | -------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------- | | [ADS](/connectors/shop-floor-connectors/ads.md) | `timestamp`, `value`, `id` | `timestamp`, `value` (`true`), `id` | | [BACnet](/connectors/shop-floor-connectors/bacnet.md) | `value`, `timestamp`, `id` | Empty message | | [EtherNet/IP](/connectors/shop-floor-connectors/ethernet-ip.md) | `id`, `value`, `timestamp` | `id`, `timestamp` | | [FOCAS](/connectors/shop-floor-connectors/focas.md) | `id`, `timestamp`, `value` | Not supported | | [Heidenhain DNC](/connectors/shop-floor-connectors/heidenhain-dnc.md) | `timestamp`, `result`, `id` (`null` if your request has none) | Same as `read` | | [HTTP/REST](/connectors/enterprise-connectors/http-rest.md#read-results) | `timestamp`, `id`, `value` | `timestamp`, `id`, `value` | | [InfluxDB and InfluxDB 3](/connectors/enterprise-connectors/influxdb.md) | The query result as an array of rows | `value` (`true`), `timestamp` | | [Kafka](/connectors/enterprise-connectors/kafka.md) | Not supported | `id` (`null` if your request has none), `timestamp`, `result` | | [Modbus/TCP](/connectors/shop-floor-connectors/modbus-tcp.md) | `timestamp`, `value`. Without `dataType`, the raw bytes. | The Modbus response | | [MQTT](/connectors/enterprise-connectors/mqtt.md) | Not supported | Empty message | | [MSSQL](/connectors/enterprise-connectors/mssql.md) | `timestamp`, `value` | `id`, `timestamp`, `value` (`true`) | | [OPC DA](/connectors/shop-floor-connectors/opc-da.md) | `timestamp`, `value` | Empty message | | [OPC UA](/connectors/shop-floor-connectors/opc-ua/opc-ua-client.md#output-format-on-reading) | The OPC UA data value. For a [batch](/connectors/shop-floor-connectors/opc-ua/opc-ua-client.md#reading-a-batch), `values`, `id`. | `id`, `timestamp`, `result` | | [Siemens SIMATIC S7](/connectors/shop-floor-connectors/siemens-simatic-s7.md) | Not supported | `id`, `timestamp`, `result` (`true`) | | [Siemens SIMATIC S7+](/connectors/shop-floor-connectors/siemens-simatic-s7-plus.md) | `timestamp`, `value`, `id` | `timestamp`, `result` (`true`) | | [Shdr](/connectors/shop-floor-connectors/shdr.md) | Not supported | Not supported | | [SINUMERIK](/connectors/shop-floor-connectors/sinumerik.md) | `id`, `timestamp`, `result` | Empty message | | [SOPAS](/connectors/shop-floor-connectors/sopas.md) | The SOPAS telegram | Empty message | | [SQL](/connectors/enterprise-connectors/sql.md) | `timestamp`, `value`, and `id` if the query has an `$id` placeholder | `id`, `timestamp`, `value` (`true`) | | [Systemstate](/connectors/enterprise-connectors/systemstate.md) | `state`, `resourceId`, `timestamp` | Not supported | | [Werma WIN Ethernet](/connectors/shop-floor-connectors/werma-win-ethernet.md) | Not supported | Not supported | An SQL `read` request can contain only the values for the placeholders of the query. Add an `id` only if the query uses an `$id` placeholder, otherwise the read fails. ### Failed Operations What Connectware publishes when an operation fails depends on the protocol: * An empty message, that is, an MQTT message without payload. Protocols that answer a successful write with an empty message answer a failed write the same way. * A result with an `error` property. `error` is a message string or an object with `message` and `code`. For HTTP/REST endpoints, ADS reads, and OPC UA writes and method calls, the [publishError](#publisherror) property controls the format. OPC UA batch endpoints always include the `error` object. * For OPC UA, a result with a Bad status code. See [OPC UA](/connectors/shop-floor-connectors/opc-ua/opc-ua-client.md#output-format-on-writing). Connectware processes requests on `/req` and `/set` while the connection of the endpoint is up. A `write` endpoint with [buffering](#buffering) also accepts messages while the connection is down and publishes the results once it writes the buffered messages. Use a timeout in your application when it waits for a result. ## Polling Interval and Subscribe When an endpoint should subscribe to some data point on the device, it should be defined with the `subscribe` operation. Some protocols support such a subscription directly (e.g., OPC UA), whereas others only support regular `polling` the data point from the Connectware side. Depending on the available choices, the actual behavior can be chosen by the properties in the [subscribe / read / write](#subscribe-read-write) section. If the endpoint is set to `polling`, there is the choice between specifying an `interval` or a `time schedule expression` for polling from the Connectware side. * An `interval` specifies the waiting time between subsequent polls. There is no guarantee on the exact time interval, only that on average the time interval should be matched. For example, if the protocol needed longer for one interval, the next one will be chosen shorter. Typical numbers for specified time intervals of 1000 milliseconds are actual intervals in the range of 950 to 1050 milliseconds, but this also strongly depends on the protocol and device behavior. * A `time schedule expression` is specified in the `cronExpression` syntax, see , for example `"0 * * * *"` for “every hour at the 00 minute, such as 00:00h, 01:00h, 02:00h, and so on. In this case, there is no guarantee on the exact time when data is received, but one polling will be triggered for each time expression match. So it can be relied on receiving 24 polling results per day if “once per hour” has been specified in the cronExpression. For any subscribe endpoint in the protocols where polling is available, you can either specify an `interval`, or a `cronExpression` (which takes precedence over the `interval` property), or neither, in which case `interval` will be used with its default value. ## Endpoint Properties | Property | Type | Required | | ---------------------------------- | ---------- | ------------ | | [protocol](#protocol) | `enum` | **Required** | | [connection](#connection) | `string` | **Required** | | [subscribe](#subscribe-read-write) | `object` | Optional\* | | [read](#subscribe-read-write) | `object` | Optional\* | | [write](#subscribe-read-write) | `object` | Optional\* | | [rules](#rules) | `object[]` | Optional | | [qos](#qos) | `integer` | Optional | | [retain](#retain) | `boolean` | Optional | | [targetState](#targetstate) | `enum` | Optional | | [topic](#topic) | `string` | Optional | | [agentName](#agentname) | `string` | Obsolete | | [buffering](#buffering) | `object` | Optional | | [inputBuffering](#inputbuffering) | `object` | Optional | | [publishError](#publisherror) | `boolean` | Optional | \* One out of [subscribe, read, and write](#subscribe-read-write) is **required**. ### protocol Identifies the protocol for which a connection should be established. * **Required** * Type: `enum` The value of this property **must** be equal to one of the following: * `Ads` * `Bacnet` * `EthernetIp` * `Focas` * `Heidenhain` * `Http` * `Influxdb` * `Influxdb3` * `Kafka` * `Modbus` * `Mqtt` * `Mssql` * `Opcda` * `Opcua` * `Openprotocol` * `S7` * `S7plus` * `Shdr` * `Sinumerik` * `Sopas` * `Sql` * `Systemstate` * `Werma` ### connection Reference to a [Cybus::Connection](/data-flows/service-commissioning-files/resources/cybus-connection.md) resource. * **Required** * Type: `string` ### subscribe / read / write * One of these is **required** * Type: `object` Depending on the [protocol](#protocol) type, this property needs the following parameters (properties) which specify the actual data address in the respective protocol: * `Ads` [Endpoint Properties](/connectors/shop-floor-connectors/ads/adsendpoint.md) * `Bacnet` [Endpoint Properties](/connectors/shop-floor-connectors/bacnet/bacnetendpoint.md) * `EthernetIp` [Endpoint Properties](/connectors/shop-floor-connectors/ethernet-ip/ethernetipendpoint.md) * `Focas` [Endpoint Properties](/connectors/shop-floor-connectors/focas/focasendpoint.md) * `Heidenhain` [Endpoint Properties](/connectors/shop-floor-connectors/heidenhain-dnc/heidenhainendpoint.md) * `Http` [Endpoint Properties](/connectors/enterprise-connectors/http-rest/httpendpoint.md) * `Influxdb` [Endpoint Properties](/connectors/enterprise-connectors/influxdb/influxdbendpoint.md) * `Influxdb3` [Endpoint Properties](/connectors/enterprise-connectors/influxdb/influxdb3endpoint.md) * `Kafka` [Endpoint Properties](/connectors/enterprise-connectors/kafka/kafkaendpoint.md) * `Modbus` [Endpoint Properties](/connectors/shop-floor-connectors/modbus-tcp/modbusendpoint.md) * `Mqtt` [Endpoint Properties](/connectors/enterprise-connectors/mqtt/mqttendpoint.md) * `Mssql` [Endpoint Properties](/connectors/enterprise-connectors/mssql/mssqlendpoint.md) * `Opcda` [Endpoint Properties](/connectors/shop-floor-connectors/opc-da/opcdaendpoint.md) * `Opcua` [Endpoint Properties](/connectors/shop-floor-connectors/opc-ua/opc-ua-client/opcuaendpoint.md) * `Openprotocol` [Endpoint Properties](/connectors/shop-floor-connectors/open-protocol/openprotocolendpoint.md) * `S7` [Endpoint Properties](/connectors/shop-floor-connectors/siemens-simatic-s7/s7endpoint.md) * `S7plus` [Endpoint Properties](/connectors/shop-floor-connectors/siemens-simatic-s7-plus/s7plusendpoint.md) * `Shdr` [Endpoint Properties](/connectors/shop-floor-connectors/shdr/shdrendpoint.md) * `Sinumerik` [Endpoint Properties](/connectors/shop-floor-connectors/sinumerik/sinumerikendpoint.md) * `Sopas` [Endpoint Properties](/connectors/shop-floor-connectors/sopas/sopasendpoint.md) * `Sql` [Endpoint Properties](/connectors/enterprise-connectors/sql/sqlendpoint.md) * `Systemstate` [Endpoint Properties](/connectors/enterprise-connectors/systemstate/systemstateendpoint.md) * `Werma` [Endpoint Properties](/connectors/shop-floor-connectors/werma-win-ethernet/wermaendpoint.md) {% hint style="warning" %} Strictly speaking, the protocol’s properties mentioned here are not properties of the endpoint but rather those of the [subscribe / read / write](#subscribe-read-write) property of the endpoint. In other words, those important properties must appear one level deeper in the YAML file: Not directly below `endpoint` but below `subscribe`/`read`/`write`, which in turn is below `endpoint`. These two different levels must not be confused. {% endhint %} ### rules * **Optional** * Type: `array` of [Rule Engine](/data-flows/rule-engine.md) You may specify rules to your payload here before sending it to the internal broker for the first time. {% hint style="warning" %} These rules will transform the raw data as received from this protocol. It affects all further steps in the processing chain. {% endhint %} ### qos MQTT Quality of Service (QoS) for the internal messaging from Endpoint to the internal MQTT broker. If this endpoint runs on an agent, setting this to `1` instead of the default `0` will activate the simple buffering of MQTT client implementations. A non-zero QoS also protects data flow during short outages of the internal broker itself, not only during agent-to-Connectware disruptions. When the broker connection is lost, the machine subscription remains active and incoming data is buffered locally until the broker is reachable again. * **Optional** * Type: `integer`, must be one of `0`, `1`, `2` * Default: `0` {% hint style="warning" %} QoS level 2 is most likely not useful in the industry context and is not recommended here. {% endhint %} ### retain Whether the last message should be retained (last-value-cached) on the internal MQTT broker. If this endpoint runs on an agent, setting this to `true` instead of the default `false` might be useful in certain applications to have some value on the topic if the agent disconnects. However, in other applications this might not make sense. * **Optional** * Type: `boolean`, must be one of `true`, `false` * Default: `false` ### targetState The state this resource should be in after startup. * **Optional** * Type: `enum`, must be one of `enabled`, `disabled` * Default: `enabled` ### topic Explicit topic name to which this endpoint address should be mapped. {% hint style="warning" %} The provided topic name is prefixed with the value of the [Cybus::MqttRoot](/data-flows/service-commissioning-files/parameters.md#global-pre-defined-parameters) global parameter. This global parameter by default has the value `services/` where `` is replaced with the actual ServiceID of the current service. Hence, in the default case the full endpoint topic will expand to: {% endhint %} `services//` See the explanation at [Cybus::MqttRoot](/data-flows/service-commissioning-files/parameters.md#global-pre-defined-parameters) if alternative topic structures are needed. Providing a custom topic and avoiding an additional mapping resource improves overall performance as the message has to travel one hop less. Endpoints with custom topics can still be mapped using a regular mapping (see [Cybus::Mapping](/data-flows/service-commissioning-files/resources/cybus-mapping.md)). * **Optional** * Type: `string` ### agentName **Obsolete**. This value is no longer being used. The agentName of the referenced connection is always used if this connection and endpoint are being used on an agent instance, separate from Connectware. ### buffering The `buffering` section can optionally switch on output data buffering on `write` endpoints. With this feature, it is possible to enable output buffering on write operations for when the connection to the device is lost, in order to avoid data loss. The buffering mechanism kicks in when a device disconnection is detected and will start buffering any incoming values. It also engages when an individual write operation fails with a timeout or write error while the connection is still reported as connected, so that transient write failures do not cause data loss. After the connection is reestablished, buffered messages will be written to the machine ("flushed"). The flushing of the buffer is implemented to handle subsequent disconnections during flushing correctly. In such a case, newly incoming values will be buffered, too. Once the connection is reestablished, the flushing will continue where it left off. {% hint style="info" %} The write buffer is owned by the endpoint. If the endpoint is destroyed as part of a service update, any messages still in its buffer at that moment are discarded and not flushed to the device. {% endhint %} By default, this feature is switched off. To enable it, the property `enabled` must be set to `true`, and most likely additional properties should be set according to the expected behavior in the actual application scenario. The supported properties of buffering are: * `enabled` (default: `false`): Whether buffering should be enabled or not when the connection to the source device is lost. * `keepOrder` (default: `true`): Whether to keep the order of messages when going into redelivery mode after an endpoint comes back online. * `burstInterval` (default: `100`): Time in milliseconds to wait between each re-publishing of buffered messages after connection is re-established. * `burstSize` (default: `1`): The number of messages to send in one burst when flushing the buffer upon re-connection. * `bufferMaxSize` (default: `100000`): The max number of messages to be buffered. Older messages are deleted when this limit is reached. * `bufferMaxAge` (default: `86400`, one day): The number of seconds the buffered data will be kept. If messages have been buffered for longer than this number of seconds, they will be discarded. {% hint style="warning" %} It is important to keep a balanced configuration of these properties to avoid potentially unwanted behavior. For example, if you configure a large `bufferMaxSize` in combination with a slow `burstInterval` and a small `burstSize`, flushing the buffer could take very long and depending on the `bufferMaxAge` messages could expire. The values should be configured based on the target device capabilities. {% endhint %} The `keepOrder` property, which is switched on by default, will keep the order of arriving messages when a flush of the buffer is in progress. This will delay newly arriving messages until all the buffered messages have been sent. For example, if we had the values `1`, `2`, `3`, `4` in the buffer, the buffer starts the flushing of values after a reconnection, and the values `5`, `6` are received in the meantime, then the machine will get the values in that exact order `1`, `2`, `3`, `4`, `5`, `6`. If this property was set to false and the same scenario is replicated, the order of arrival of the new values is unspecified and the end result would be an interleaved set of values, for example: `1`, `5`, `2`, `3`, `6`, `4`. ### inputBuffering The input data to each endpoint can optionally be managed through an individual input buffer (also called `input queue`) to establish fine-grained control for high data rate behavior. By default, this input buffering is disabled and instead all input data is handled on the global event queue, which works fine as long as there is no risk of out-of-memory exceptions due to unexpected slow data processing or forwarding. When enabling the individual input buffer, the buffer properties determine the behavior in situations when the input buffer is filling up. The buffer is filling up when the message arrival rate is larger than the processing data rate or the forwarding (publishing) data rate. Or, in other words, the input buffer is filling up if the messages arrive faster than how they can be processed or be forwarded (published). If this situation happens for longer time durations, the input buffer will reach its configured capacity limits and arriving messages will be dropped, so that the system will not run into an uncontrollable out-of-memory exception. This is a fundamental and unavoidable property of distributed systems due to its finite resources. But the actual behavior of the input buffer can be adapted to the actual application scenario by setting the properties in the `inputBuffering` section (optional). Supported properties are (all optional): * `enabled` (type: `boolean`, default: `false`): Enable or disable input buffering. * `maxInputBufferSize` (type: `integer`, default: `5000`): Maximum number of input messages that are queued in the input buffer. Exceeding messages will be discarded. Adjust this to a higher value if you are handling bursty traffic. * `maxConcurrentMessages` (type: `integer`, default: `2`): Maximum number of concurrently processed messages as long as the input buffer queue is non-empty. * `waitingTimeOnEmptyQueue` (type: `integer`, default: `10`): Waiting time in milliseconds after the input buffer queue ran empty and before checking again for newly queued input messages. Regardless of this value, on non-empty input buffer queue all messages will be processed without waiting time in between until the queue is empty again. ### publishError Whether Connectware publishes a failed `read` or `write` operation on the `/res` topic as an `error` object with `message` and `code`. Set it to `true` when you use the endpoint in a [FlowSync](/data-flows/flowsync.md) flow. * **Optional** * Type: `boolean`, must be one of `true`, `false` * Default: `false` `publishError` has an effect on HTTP/REST endpoints, on ADS reads, and on OPC UA writes and method calls: | Operation | `publishError: false` | `publishError: true` | | ------------------------------- | --------------------------------------------------------------- | ------------------------------------------- | | HTTP/REST `read` | The result is an empty message. | The result contains an `error` object. | | HTTP/REST `write` | The result contains `error` as a string with the message. | The result contains an `error` object. | | ADS `read` | The result is an empty message. | The result contains an `error` object. | | OPC UA `write` and method calls | The result contains the status code in `result` and no `error`. | The result also contains an `error` object. | An OPC UA [batch endpoint](/connectors/shop-floor-connectors/opc-ua/opc-ua-client.md#batch-reading-and-writing) ignores `publishError`. It always includes the `error` object when it rejects a message or aborts a batch. The following result of an HTTP/REST `read` endpoint contains an `error` object: ```json { "timestamp": 1629351968526, "id": "read-1", "error": { "message": "GET request on /foo failed with status: 404 - Not Found", "code": 404 } } ``` For the complete result formats, see [HTTP/REST](/connectors/enterprise-connectors/http-rest.md#read-results) and [OPC UA](/connectors/shop-floor-connectors/opc-ua/opc-ua-client.md#output-format-on-writing). ## Examples ### BACnet {% code lineNumbers="true" %} ```yaml bacnetSubscribe: type: Cybus::Endpoint properties: protocol: Bacnet connection: !ref bacnetConnection subscribe: objectType: analog-input objectInstance: 2796204 interval: 1000 # This subscribes to a Bacnet analog input of object instance 2796204 ``` {% endcode %} ### OPC UA {% code lineNumbers="true" %} ```yaml opcuaSubscribeToCurrentServerTime: type: Cybus::Endpoint properties: protocol: Opcua connection: !ref opcuaConnection subscribe: nodeId: i=2258 # This subscribes to the OPC UA server node that publishes the current time ``` {% endcode %} ### MQTT with Write Buffering Enabled {% code lineNumbers="true" %} ```yaml writeEndpoint: type: Cybus::Endpoint properties: protocol: Mqtt connection: !ref mqttConnection buffering: enabled: true keepOrder: true burstInterval: 10 burstSize: 100 bufferMaxSize: 20000 bufferMaxAge: 5000 write: topic: test/write # This configures a write endpoint which will buffer up to 20000 messages # if the connection is lost and will publish 100 messages every 10 milliseconds # when the connection is reestablished. New incoming messages will be published # only when the originally buffered items were all published ``` {% endcode %} --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation by asking a question. Perform an HTTP GET request on the following URL with the `ask` and `goal` query parameters: ``` GET https://docs.cybus.io/data-flows/service-commissioning-files/resources/cybus-endpoint.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is what the user is ultimately trying to achieve, the reason they need the answer. Sharing it helps GitBook give you a better, more relevant answer. A goal is most helpful when it describes the outcome the user wants rather than restating the question. For example, with `ask=how do I create an API token`, a goal like `automate deployments from our CI pipeline` lets GitBook tailor the answer to that use case. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.