SQL

Connect Connectware to MariaDB and PostgreSQL databases for read, write, and subscribe operations with the SQL connector.

The SQL connector enables Connectware to communicate with relational databases. This implementation uses Sequelize (https://sequelize.org/), a Node.js ORM (Object-Relational Mapping) tool, to provide read, write, and subscribe capabilities for database operations.

Supported Databases

• MariaDB

• PostgreSQL

• Oracle

Usage

In general, the implementation works by specifying a query or query template.

The following examples use a sample table called people that looks like this:

id	name	lastname	gender
1	Alice	Miller	female
2	Bob	Jones	male

Connection

To connect to a database, you must provide a URL when specifying the Cybus::Connection resource:

<schema>://<user>:<password>@<host>:<port>/<database>

For example, to connect to a MariaDB database use this:

'mariadb://johndoe:my-secret-pw@localhost:3306/test'

To connect to a Postgres database use this:

'postgres://johndoe:my-secret-pw@localhost/test'

To connect to an Oracle database use this:

'oracle://johndoe:my-secret-pw@localhost:1521/ORCLPDB1'

The path segment after the host is the Oracle service name, not a database name. For TLS-encrypted Oracle connections, see Oracle Connections and TLS.

We recommend defining these properties as parameters in the service commissioning file and reference them in the connection settings using !ref, so that the actual value can be edited at deployment time. See Service Commissioning File Example below.

For more information, see Connection Properties.

Reading Data

To read data from a database, an endpoint has to be defined with either read or subscribe properties. Subscribe works by defining a polling interval, hence the query will be executed on a regular basis. Read is executed each time an MQTT message is sent to the respective endpoint topic with the /req (request) suffix, where the result is sent to the endpoint topic with the /res suffix. The result of the query is provided in JSON format on the MQTT broker.

Example endpoint definition:

sqlQuery1:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref sqlConnection
    subscribe:
      query: 'SELECT * FROM people'
      interval: 2000

This endpoint will execute the given query and return the data as MQTT messages like in the following example. If no rows are returned, you will receive an empty array ([]) as a value.

{
  "timestamp": 1231782312, // unix timestamp in ms
  "value": [
    {
      "id": 1,
      "name": "Alice",
      "lastname": "Miller",
      "gender": "female"
    },
    {
      "id": 2,
      "name": "Bob",
      "lastname": "Jones",
      "gender": "male"
    }
  ]
}

Placeholder Syntax

The SQL query definition can be defined as a template string containing placeholders in the WHERE clause. In the template, the dollar sign character $ followed by an identifier is used to denote placeholders. The placeholders will be replaced by the values from the payload of the JSON message received via MQTT. If no placeholders are defined, the query will be executed as-is.

Placeholder naming rules:

The regular expression for placeholder identifiers is $[a-zA-Z0-9_]+. This means:

• Placeholders start with a single dollar sign $

• Followed by an identifier consisting of ASCII letters, numbers, and underscores

• Examples: $name, $user_id, $value123

Special cases:

• Multiple dollar signs (e.g., $$VERSION) are not interpreted as placeholders and will be passed unchanged to the SQL query.

• A single dollar sign followed by whitespace or punctuation (e.g., $ or $.) is not interpreted as a placeholder.

• To use a single $ that matches the pattern above in the SQL query, define it as a placeholder and send the desired string as a value in the message payload.

Usage requirements:

If the SQL query contains placeholder definitions, all their names must exist in the message payload, otherwise an error will be logged and the message will be ignored. The value of the placeholders must have the right data format matching the target schema of the database.

Example endpoint definition:

sqlQuery1:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref sqlConnection
    read:
      query: 'SELECT * FROM people WHERE lastname = $lastname'

Sending a message to the /req topic of this endpoint with the following payload:

{ "lastname": "Miller" }

will return results filtered based on the where clause configured:

{
  "timestamp": 1231792312, // unix timestamp in ms
  "value": [
    {
      "id": 1,
      "name": "Alice",
      "lastname": "Miller",
      "gender": "female"
    }
  ]
}

Output Format on Read

When data is read from SQL, results are published to the /res topic of the endpoint. The output message is an object with two properties:

• timestamp: is the unix timestamp, in milliseconds, of when the read was executed

• value: is an array of results as returned by the SQL query

Writing Data

To write data to the database, an endpoint with write properties has to be defined. This endpoint needs the definition of an SQL query with a query template containing variables. The variables will be filled by the values of the payload of the JSON message received via MQTT.

In the SQL query definition, the query syntax is used as a template string containing placeholders. In the template, the dollar sign $ followed by an identifier is used to denote placeholders (see Placeholder Syntax for details). The placeholders will be replaced by the values from the payload of the JSON message received via MQTT.

All specified template variables must exist in the payload and must have the right target format matching the schema of the database.

Example endpoint definition:

sqlQuery1:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref sqlConnection
    write:
      query: 'INSERT INTO people (name, lastname, gender) VALUES ($name, $lastname, $gender)'

When using bulk insert you need to specify the endpoint like this:

sqlQuery2:
  type: Cybus::Endpoint
  properties:
    protocol: Sql
    connection: !ref sqlConnection
    write:
      query: 'INSERT INTO people (name, lastname, gender) VALUES'
      queryValues: '($name, $lastname, $gender)'

To write data, you must send an MQTT message like the following to the /set topic of the endpoint:

{
  "name": "Alice",
  "lastname": "Miller",
  "gender": "female"
}

Alternatively, you can also send multiple rows into a single message for performance reasons like this:

When using this method of insertion make sure all rows have the same amount of columns.

You will also need to specify the parameter queryValues in the endpoint definition.

[
  { "name": "Alice", "lastname": "Miller", "gender": "female" },
  { "name": "John", "lastname": "Clark", "gender": "male" }
]

The SQL connection on the Connectware side does not perform any data validation against the database schema. The senders of the MQTT messages themselves must ensure to send the data in the correct format.

Output Format on Write

When data is written to an SQL endpoint, a message is published to the /res topic of the endpoint. The output message is an object with two properties:

• timestamp: is the unix timestamp, in milliseconds, of when the write was executed

• value: is set to true when the write was successful

• Connection Properties

• Endpoint Properties

Oracle Connections and TLS

Reading, writing, placeholders, and output formats work the same for Oracle as for the other dialects. This section covers what is specific to Oracle: the underlying driver mode and the TLS wallet configuration.

Connectware connects to Oracle Database using the node-oracledb driver in Thin mode, a pure JavaScript implementation that does not require the Oracle Client libraries to be installed. Native Network Encryption (NNE) is not supported; encrypted connections must use TLS.

Encrypted Connections with a Wallet

Oracle accepts TLS credentials as a wallet, a bundle containing an X.509 client certificate, its private key, and the trusted CA or server certificate. You supply the wallet to the connection through the walletContent and walletPassword properties.

Providing walletContent is what switches the connection to TLS. Without it, the connection is plain and unencrypted.

Property	Required	Description
walletContent	yes	Base64-encoded PEM bundle containing the private key, the client certificate, and the trusted CA or server certificate, in any order. Setting this property enables TLS.
walletPassword	no	Passphrase for the private key inside the wallet. Only required when the private key is encrypted (-----BEGIN ENCRYPTED PRIVATE KEY-----).

The wallet must include a private key, even for one-way TLS. A server certificate on its own is not sufficient.

Oracle listens on port 2484 for TLS (TCPS) connections and on 1521 for plain connections. Use the port that matches your database configuration in the connection URL.

The trustAllCertificates property accepts self-signed certificates and relaxes the server certificate hostname (DN) match. Set it to false in production when the server certificate is issued for the exact host you connect to.

Creating the Wallet Bundle

If your Oracle Database provides a PKCS#12 wallet (ewallet.p12), convert it to an unencrypted PEM bundle and Base64-encode the result:

# Convert the PKCS#12 wallet to an unencrypted PEM bundle (key + cert + CA)
openssl pkcs12 -in ewallet.p12 -out bundle.pem -nodes -passin pass:${WALLET_PASSWORD}

# Base64-encode the bundle for walletContent
base64 -w0 bundle.pem

• ${WALLET_PASSWORD} is the passphrase of the source PKCS#12 wallet, not the value you pass to walletPassword.

• The -nodes flag produces an unencrypted private key, so you do not need to set walletPassword on the connection.

Service Commissioning File Example for Oracle with TLS

sql-oracle-tls-example.yml

2KB

Open

sql-oracle-tls-example.yml

---
description: >
  Sample Oracle SQL service commissioning file with TLS (TCPS) enabled via a Base64-encoded wallet.

metadata:
  name: Sample Oracle SQL service (TLS)
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  oracleUser:
    type: string
    default: myuser
  oraclePassword:
    type: string
    default: my-secret-pw
  oracleHost:
    type: string
    default: oracle.example.com
  oraclePort:
    type: integer
    # 2484 is the standard Oracle TCPS (TLS) port. Use 1521 for plain, unencrypted connections.
    default: 2484
  oracleServiceName:
    type: string
    default: ORCLPDB1
  # Base64-encoded PEM bundle containing the private key, client certificate, and trusted CA/server certificate.
  # Providing walletContent switches the connection to TLS.
  walletContent:
    type: string
  # Only required if the private key inside the wallet is encrypted.
  walletPassword:
    type: string
    default: ''

resources:
  oracleConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'oracle://${oracleUser}:${oraclePassword}@${oracleHost}:${oraclePort}/${oracleServiceName}'
        # Accept self-signed certificates and relax the server certificate hostname (DN) match.
        # Set to false in production when the server certificate is issued for the exact host.
        trustAllCertificates: true
        walletContent: !ref walletContent
        walletPassword: !ref walletPassword

  oracleQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref oracleConnection
      subscribe:
        query: 'SELECT machine_id, temperature FROM sensor_readings'
        interval: 5000

  mapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref oracleQuery
          publish:
            topic: 'enterprise/oracle/sensor-readings'

Service Commissioning File Example

sql-example.yml

2KB

Open

sql-example.yml

---
description: >
  Sample SQL service commissioning file

metadata:
  name: Sample SQL service
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  sqlURL:
    type: string
    default: 'mariadb://root:my-secret-pw@localhost:3306/mysql'
  initialReconnectDelay:
    type: integer
    default: 1000
  maxReconnectDelay:
    type: integer
    default: 30000
  factorReconnectDelay:
    type: integer
    default: 2

resources:
  sqlConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !ref sqlURL
        connectionStrategy:
          initialDelay: !ref initialReconnectDelay
          maxDelay: !ref maxReconnectDelay
          incrementFactor: !ref factorReconnectDelay

  sqlQuery1:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref sqlConnection
      subscribe:
        query: 'SELECT User FROM user'
        # Be very careful with this setting. A low value might overload
        # the database (the unit is milliseconds!).
        interval: 2000

  sqlQuery2:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref sqlConnection
      write:
        # Here we use the placeholder $someValue. The protocol driver will insert
        # the value from the input JSON message under the key someValue
        query: 'INSERT INTO test (id, some_column) VALUES ($id, $someValue)'

  mapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref sqlQuery1
          publish:
            topic: 'database-users'
        - subscribe:
            topic: 'database-insert-test'
          publish:
            endpoint: !ref sqlQuery2