MessageBus

The MessageBus interface is used for communication between different components, for example to synchronize between endpoints. Therefore, the MessageBus is primarily designed for internal use but as it might also be useful for some applications and scenarios there might be implementations that expose the MessageBus to the outside world.

Events

The MessageBus works according to the publish/subscribe principle based on different types of events or event messages (which are subclasses of the abstract class EventMessage). Subscriptions are made to a kind of event, i.e. a subclass of EventMessage or even EventMessage itself (to receive all events). When subscribing to a class, all events of this class or any subclass are received.

This is the class hierarchy of available event classes/types

• EventMessage (abstract): Superclass for all events, payload: a Reference to the subject element

  • AccessEventMessage (abstract): Superclass for all types of access-based events

    • ReadEventMessage (abstract): Superclass for all types of read-events, triggered each time an element is read via API

      • ElementReadEventMessage: Triggered when a Referable is read via API, payload: the referable (serialized according to the request, i.e. using the requested SerializationModifier)

      • ValueReadEventMessage: Triggered when the value of an element is read via API, payload: the element value

    • ExecuteEventMessage (abstract): Superclass for all events related to executing operations

      • OperationInvokeEventMessage: Triggered when an operation is invoked/started, payload: input and inoutput parameters

      • OperationFinishEventMessage: Triggered when an operation is finished, payload: output and inoutput parameters

  • ChangeEventMessage (abstract): Superclass for all types of changes

    • ElementChangeEventMessage (abstract): Superclass for all types of structural changes, payload: the updated element

      • ElementCreateEventMessage: Triggered when an element is created

      • ElementDeleteEventMessage: Triggered when an element is deleted

      • ElementUpdateEventMessage: Triggered when an element is updated

    • ValueChangeEventMessage: Triggered when the value of an element is updated, payload: old value, new value

  • ErrorEventMessage: Triggered when an error occurred, payload: message, error level (INFO, WARN, ERROR)

Internal

This is the default implementation of the MessageBus interface which is implemented using Java method calls. Therefore, it can only be accessed from code when FA³ST Service is used as an embedded library.

Configuration

This implementation does not offer any configuration properties.

Example configuration for Internal MessageBus.

{
	"messageBus": {
		"@class": "de.fraunhofer.iosb.ilt.faaast.service.messagebus.internal.MessageBusInternal"
	},
	//...
}

MQTT

This implementation of the MessageBus interface publishes messages via MQTT either by hosting its own MQTT server or by using an externally hosted one.

Topics & Payload

Each message type is published on its own topic in the form of [topicPrefix]/[className], e.g. events/ValueChangeEventMessage. The payload is a JSON serialization of the corresponding Java class with the following base structure

JSON structure of serialized MessageBus events.

{
	"@type": "[event type]",
	"element": { 
		// [default JSON serialization of Reference] 
	},
	// [event-specific properties]
}

An example ValueChangeEvent might look like this:

JSON serialization of an example ValueChangeEvent.

{
    "@type": "ValueChangeEvent",
    "element": {
        "keys": [
            {
                "type": "Submodel",
                "value": "http://example.org/submodel"
            },
            {
                "type": "Property",
                "value": "property"
            }
    ] },
    "oldValue": {
        "modelType": "Property",
        "dataType": "xs:int",
        "value": 0
    },
    "newValue": {
        "modelType": "Property",
        "dataType": "xs:int",
        "value": 1
    }
}

For deserialization of events the class JsonEventDeserializer in module dataformat-json can be used.

Configuration

Configuration properties of MQTT MessageBus.

Name	Allowed Value	Description	Default Value
clientCertificate (optional)	CertificateInfo	The client certificate to use. If not set, SSL will be disabled.
clientId (optional)	String	ClientId to use when connecting to the MQTT server.	FAST MQTT MessageBus
host (optional)	String	The host name of the MQTT server without prefix, e.g., 192.168.0.1.	0.0.0.0
password (optional)	String	Password used to connect to the MQTT server.
port (optional)	Integer	The port to use for TCP communication.	1883
serverCertificate (optional)	CertificateInfo	The server certificate to use. If not set, SSL will be disabled.
sslPort (optional)	Integer	The port to use for secure TCP communication.	8883
sslWebsocketPort (optional)	Integer	The port to use for secure websocket communication.	443
topicPrefix (optional)	String	Prefix to use for the topic names.	events/
useInternalServer (optional)	Boolean	If true, FA³ST Service starts its own MQTT server. If false, FA³ST Service uses external MQTT server.	true
username (optional)	String	Username used to connect to the MQTT server.
users (optional)	Map<String, String>	Map of usernames and passwords of users that are allowed to connect to the MQTT server. This is only used when useInternalServer is true	empty list
useWebsocket (optional)	Boolean	If true uses websocket, otherwise TCP.	false
websocketPort (optional)	Integer	The port to use for TCP communication	9001

Example configuration for MQTT MessageBus.

{
	"messageBus": {
		"@class": "de.fraunhofer.iosb.ilt.faaast.service.messagebus.mqtt.MessageBusMqtt",
		"useInternalServer": true,
		"port": 1883,
		"sslPort": 8883,
		"host": "localhost",
		"websocketPort": 9001,
		"sslWebsocketPort": 443,
		"serverCertificate": {
			"keyStoreType": "PKCS12",
			"keyStorePath": "C:\faaast\MyKeyStore.p12",
			"keyStorePassword": "changeit",
			"keyAlias": "server-key",
			"keyPassword": "changeit"
		},
		"clientCertificate": {
			"keyStoreType": "PKCS12",
			"keyStorePath": "C:\faaast\MyKeyStore.p12",
			"keyStorePassword": "changeit",
			"keyAlias": "client-key",
			"keyPassword": "changeit"
		},
		"users": {
			"user1": "password1"
		},
		"username": "messagebus-user",
		"password": "messagebus-password",
		"clientId": "CustomClientId",
		"topicPrefix": "faaast/events/"
	},
	//...
}

CloudEvents

This implementation of the MessageBus interface publishes CloudEvent messages via MQTT by using an externally hosted MQTT broker.

Topics & Payload

Each message is published under the topic defined in [topicPrefix]. The payload is a JSON serialization of a CloudEvent as specified in the async-aas specification: https://factory-x-contributions.github.io/async-aas-helm

Note: To modify the URL in a CloudEvent’s source field, HttpEndpointConfig.hostname needs to be configured. Else, http(s)://(localhost|127.0.0.1|local-ip-address):(port) will be used, depending on the rest of the HttpEndpoint’s config and your system. If multiple HttpEndpoints are defined, the MessageBus will try to select one with a non-local hostname.

Note: MessageBusCloudEvents requires the definition of a HttpEndpoint in the FA³ST configuration.

Configuration

Configuration properties of CloudEvents MessageBus.

Name	Allowed Value	Description	Default Value
host	String	The host name of the MQTT server with scheme, port and path segments if relevant.
clientId (optional)	String	ClientId to use when connecting to the MQTT server. This clientId will only be used to identify as a MQTT publisher, not for oauth-flows or as a username.	A randomized identifier
username (optional)	String	Username used to connect to the MQTT server.
password (optional)	String	Password used to connect to the MQTT server.
identityProviderUrl (optional)	String	Oauth2 IdP URL. Obtained tokens are placed as the MQTT broker password, the username is obtained from the username config. Token refresh happens automatically.
oauth2ClientId (optional)	String	Oauth2 client id to use when retrieving an oauth2-token from the IdP.
oauth2ClientSecret (optional)	String	Oauth2 client secret to use when retrieving an oauth2-token from the IdP.
topicPrefix (optional)	String	Prefix to use for the topic names.	noauth
slimEvents (optional)	boolean	Whether the full referable is sent in a CloudEvent’s data-field. If true, the data-field will be empty.	true
eventTypePrefix (optional)	String	Prefix to use in the CloudEvents type-field. Should end with a “.”.	io.admin-shell.events.v1.
dataSchemaPrefix (optional)	String	Prefix to use in the CloudEvents dataschema-field. Should end with a “/”.	https://api.swaggerhub.com/domains/Plattform_i40/Part1-MetaModel-Schemas/V3.1.0#/components/schemas/
clientCertificate (optional)	CertificateInfo	The client certificate to use. If not set, SSL will be disabled.

Example configuration for CloudEvents MessageBus.

{
	"messageBus": {
		"@class": "de.fraunhofer.iosb.ilt.faaast.service.messagebus.cloudevents.MessageBusCloudEvents",
		"host": "tcp://localhost:1883",
		"clientCertificate": {
			"keyStoreType": "PKCS12",
			"keyStorePath": "C:\faaast\MyKeyStore.p12",
			"keyStorePassword": "changeit",
			"keyAlias": "client-key",
			"keyPassword": "changeit"
		},
		"username": "broker-user",
		"password": "broker-password",
		"identityProviderUrl": "http://localhost:12345/realms/fa3st/protocol/openid-connect/token",
		"oauth2ClientId": "my-keycloak-clientid",
		"oauth2ClientSecret": "my-keycloak-clientsecret",
		"clientId": "FA³ST Service client",
		"topicPrefix": "noauth",
		"slimEvents": true,
		"eventTypePrefix": "io.admin-shell.events.v1.",
		"dataSchemaPrefix": "https://api.swaggerhub.com/domains/Plattform_i40/Part1-MetaModel-Schemas/V3.1.0#/components/schemas/"
	},
	//...
}