IIoT (Industrial IoT) architectures are typically distributed and asynchronous, with communication being event-driven, such as the publication (and corresponding subscription) of messages. These asynchronous architectures enhance scalability and tolerance to changes, but raise interoperability issues as the explicit knowledge of the internal structure of the messages and their categorization (topics) is diluted among the elements of the architecture. 

In fact, this was also a problem for REST APIs, until the industry came together and proposed a standard way to define the structure and schema of synchronous APIs: OpenAPI (derived from Swagger). For asynchronous architectures, and inspired by OpenAPI, the AsyncAPI has appeared recently:

AsyncAPI provides a specification that allows you to define Message-Driven APIs in a machine-readable format. It's protocol-agnostic, so you can use it for APIs that work over Kafka, MQTT, AMQP, WebSockets, STOMP, etc. The spec is very similar to OpenAPI/Swagger so, if you're familiar with it, AsyncAPI should be easy for you.

In AsyncAPI, the specifications of an API can be defined in YAML or JSON, which allows specifying, for example, the message brokers, the topics of interest, or the different message formats associated with each one of the topics, among other aspects. AsyncAPI is, however, in the early stages of development, and the AsyncAPI tool market is underdeveloped, mainly limited to the generation of documentation to be consumed by humans.

Similarly to what we've done for OpenAPI (see our API Composer or our API Discoverer), we believe a model-based approach would facilitate the modeling of AsyncAPI specifications and the development of Message-Driven APIs from them. 

The overall view of our approach is illustrated in the following figure.

The AsyncAPI toolkit is available on GitHub, make sure you star/watch it to follow its evolution!.

Importing / Modeling an AsyncAPI specification

First, based on the AsyncAPI specification, we created an Xtext grammar. From this grammar, an Ecore metamodel is automatically derived, together with a set of editors and Eclipse-based tools. These editors allow creating JSON-based specifications of message-driven APIs using AsyncAPI. Specifications created using these editors are automatically parsed and reified as instances of the AsyncAPI metamodel.

Generating code to easily process messages from an AsyncAPI specification

Additionally, the prototype is able to generate Java code supporting the creation and serialization of JSON-based message payloads according to the modeled AsyncAPI, including nested JSON objects. No support for arrays is provided yet at this point however. The excerpt below shows an example of an AsyncAPI specification supported by the prototype:

{
  "asyncapi": "1.2.0",
  "info": {
    "title": "Sample AsyncAPI specification",
    "version": "0.1.0",
  },
  "servers": [
    {
      "url": "broker.url:{port}",
      "scheme": "mqtt",
      "description": "This is an example description",
      "variables": {
        "port": {
          "default": "1883",
          "enum": [ "1883", "8883" ]
        }
      }
    }
  ],
  "topics": {
    "messages/device2controller": {
      "publish": { "$ref" : "#/components/messages/request“ }
    }
  }
},
"components": {
  "schemas": {
    "protocol_version": {
      "title": "Protocol version",
      "type": "integer",
      "default": 2,
      "x-friendly-name": "ProtocolVersion"
    },
    "id": {
      "title": "ID",
      "type": "string",
      "format": "XXXXXX YY ZZZZZZ W"
    },
    "status": {
      "title": "Status",
      "type": "string",
      "enum": ["OK", "ERROR"],
      "x-friendly-name" : "Status"
    },
    "environment": {
      "title": "Environment",
      "type": "string",
      "enum": ["DEV", "STAG","PROD" ],
      "x-friendly-name" : "Environment"
    }
  },
  "messages" : {
    "request" : {
      "summary" : "Request connectivity.",
      "description": "Request connectivity when status changes",
      "payload": {
        "type": "object",
        "properties": {
          "P": { "$ref": "#/components/schemas/protocol_version" },
          "ID": { "$ref": "#/components/schemas/id" },
          "E": { "$ref": "#/components/schemas/environment" },
          "M": {
            "x-friendly-name" : "Message",
            "properties": {
              "S": { "$ref": "#/components/schemas/status" },
              "C": {
                "title": "Content",
                "type": "string",
                "x-friendly-name": "Content"
              }
            }
          }
        }
      }
    }
  }
}

A specification like the above, allows generating messages as follows:

package tests;
import messages.device2controller.Request;
import messages.device2controller.Request.Payload.Environment;
import messages.device2controller.Request.Payload.Message;
import messages.device2controller.Request.Payload.PayloadBuilder;
import messages.device2controller.Request.Payload.Message.Status;

public class Test {

  public static void main(String[] args) {

    PayloadBuilder builder = Request.payloadBuilder();
 
    Request.Payload payload = builder
      .withProtocolVersion(2)
      .withEnvironment(Environment.DEV)
      .withID("id")
      .withMessage(
        Message.newBuilder()
          .withStatus(Status.OK)
          .withContent("Content")
          .build()
      ).build();
 
    System.out.println(payload.toJson(true));
    System.out.println(Request.Payload.fromJson(payload.toJson()).toJson(true));
  }
}

The code generated by our toolkit also allows to easily publish the messages built as explained above, and to subscribe to them using the servers configured in the AsyncAPI specification. Check our online documentation for an example!

Generating a new AsyncAPI from an Ecore model

Until now, we assumed that either you already had an AsyncAPI file to import or you would be using our AsyncAPI editor to create one. In fact, there is a third alternative: take an existing Ecore model you already have available and generate an skeleton AsyncAPI specification from it.

The generator will create a reusable JSON Schema for each domain class. Channels will be created out of annotated EClasses. Moreover, hosts information can also be specified via EAnnotations (more details).