Kafka asyncapi client

incubator/binding-asyncapi.spec/NOTICE

@@ -17,6 +17,7 @@ This project includes:
   Jakarta JSON Processing API under Eclipse Public License 2.0 or GNU General Public License, version 2 with the GNU Classpath Exception
   org.leadpony.justify under The Apache Software License, Version 2.0
   zilla::specs::binding-http.spec under The Apache Software License, Version 2.0
+  zilla::specs::binding-kafka.spec under The Apache Software License, Version 2.0
   zilla::specs::binding-mqtt.spec under The Apache Software License, Version 2.0
   zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
   zilla::specs::engine.spec under The Apache Software License, Version 2.0

incubator/binding-asyncapi.spec/pom.xml

@@ -51,6 +51,11 @@
       <artifactId>binding-http.spec</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-kafka.spec</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
@@ -88,7 +93,7 @@
         <artifactId>flyweight-maven-plugin</artifactId>
         <version>${project.version}</version>
         <configuration>
-          <scopeNames>core http mqtt asyncapi</scopeNames>
+          <scopeNames>core http mqtt kafka asyncapi</scopeNames>
           <packageName>io.aklivity.zilla.specs.binding.asyncapi.internal.types</packageName>
         </configuration>
         <executions>

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/client.secure.yaml → incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/client.kafka.sasl.yaml

@@ -26,16 +26,21 @@ bindings:
     vault: test0
     options:
       specs:
-        - mqtt/asyncapi.yaml
+        - kafka/asyncapi.yaml
       tcp:
         host: localhost
         port:
-          - 7183
+          - 9092
       tls:
         trust:
           - serverca
         trustcacerts: true
         sni:
-          - mqtt.example.net
+          - kafka.example.net
         alpn:
-          - mqtt
+          - kafka
+      kafka:
+        sasl:
+          mechanism: plain
+          username: username
+          password: password

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/client.yaml → incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/client.kafka.yaml

@@ -22,8 +22,8 @@ bindings:
     kind: client
     options:
       specs:
-        - mqtt/asyncapi.yaml
+        - kafka/asyncapi.yaml
       tcp:
         host: localhost
         port:
-          - 7183
+          - 9092

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/kafka/asyncapi.yaml

@@ -0,0 +1,199 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+asyncapi: 3.0.0
+info:
+  title: Streetlights Kafka API
+  version: 1.0.0
+  description: "The Smartylighting Streetlights API allows you to remotely manage the city lights.\n\n### Check out its awesome features:\n\n* Turn a specific streetlight on/off \U0001F303\n* Dim a specific streetlight \U0001F60E\n* Receive real-time information about environmental lighting conditions \U0001F4C8\n"
+  license:
+    name: Apache 2.0
+    url: 'https://www.apache.org/licenses/LICENSE-2.0'
+defaultContentType: application/json
+servers:
+  scram-connections:
+    host: 'test.mykafkacluster.org:18092'
+    protocol: kafka-secure
+    description: Test broker secured with scramSha256
+    security:
+      - $ref: '#/components/securitySchemes/saslScram'
+    tags:
+      - name: 'env:test-scram'
+        description: >-
+          This environment is meant for running internal tests through
+          scramSha256
+      - name: 'kind:remote'
+        description: This server is a remote server. Not exposed by the application
+      - name: 'visibility:private'
+        description: This resource is private and only available to certain users
+channels:
+  lightingMeasured:
+    address: 'smartylighting.streetlights.1.0.event.{streetlightId}.lighting.measured'
+    messages:
+      lightMeasured:
+        $ref: '#/components/messages/lightMeasured'
+    description: The topic on which measured values may be produced and consumed.
+    parameters:
+      streetlightId:
+        $ref: '#/components/parameters/streetlightId'
+  lightTurnOn:
+    address: 'smartylighting.streetlights.1.0.action.{streetlightId}.turn.on'
+    messages:
+      turnOn:
+        $ref: '#/components/messages/turnOnOff'
+    parameters:
+      streetlightId:
+        $ref: '#/components/parameters/streetlightId'
+  lightTurnOff:
+    address: 'smartylighting.streetlights.1.0.action.{streetlightId}.turn.off'
+    messages:
+      turnOff:
+        $ref: '#/components/messages/turnOnOff'
+    parameters:
+      streetlightId:
+        $ref: '#/components/parameters/streetlightId'
+  lightsDim:
+    address: 'smartylighting.streetlights.1.0.action.{streetlightId}.dim'
+    messages:
+      dimLight:
+        $ref: '#/components/messages/dimLight'
+    parameters:
+      streetlightId:
+        $ref: '#/components/parameters/streetlightId'
+operations:
+  receiveLightMeasurement:
+    action: receive
+    channel:
+      $ref: '#/channels/lightingMeasured'
+    summary: >-
+      Inform about environmental lighting conditions of a particular
+      streetlight.
+    traits:
+      - $ref: '#/components/operationTraits/kafka'
+    messages:
+      - $ref: '#/channels/lightingMeasured/messages/lightMeasured'
+  turnOn:
+    action: send
+    channel:
+      $ref: '#/channels/lightTurnOn'
+    traits:
+      - $ref: '#/components/operationTraits/kafka'
+    messages:
+      - $ref: '#/channels/lightTurnOn/messages/turnOn'
+  turnOff:
+    action: send
+    channel:
+      $ref: '#/channels/lightTurnOff'
+    traits:
+      - $ref: '#/components/operationTraits/kafka'
+    messages:
+      - $ref: '#/channels/lightTurnOff/messages/turnOff'
+  dimLight:
+    action: send
+    channel:
+      $ref: '#/channels/lightsDim'
+    traits:
+      - $ref: '#/components/operationTraits/kafka'
+    messages:
+      - $ref: '#/channels/lightsDim/messages/dimLight'
+components:
+  messages:
+    lightMeasured:
+      name: lightMeasured
+      title: Light measured
+      summary: >-
+        Inform about environmental lighting conditions of a particular
+        streetlight.
+      contentType: application/json
+      traits:
+        - $ref: '#/components/messageTraits/commonHeaders'
+      payload:
+        $ref: '#/components/schemas/lightMeasuredPayload'
+    turnOnOff:
+      name: turnOnOff
+      title: Turn on/off
+      summary: Command a particular streetlight to turn the lights on or off.
+      traits:
+        - $ref: '#/components/messageTraits/commonHeaders'
+      payload:
+        $ref: '#/components/schemas/turnOnOffPayload'
+    dimLight:
+      name: dimLight
+      title: Dim light
+      summary: Command a particular streetlight to dim the lights.
+      traits:
+        - $ref: '#/components/messageTraits/commonHeaders'
+      payload:
+        $ref: '#/components/schemas/dimLightPayload'
+  schemas:
+    lightMeasuredPayload:
+      type: object
+      properties:
+        lumens:
+          type: integer
+          minimum: 0
+          description: Light intensity measured in lumens.
+        sentAt:
+          $ref: '#/components/schemas/sentAt'
+    turnOnOffPayload:
+      type: object
+      properties:
+        command:
+          type: string
+          enum:
+            - 'on'
+            - 'off'
+          description: Whether to turn on or off the light.
+        sentAt:
+          $ref: '#/components/schemas/sentAt'
+    dimLightPayload:
+      type: object
+      properties:
+        percentage:
+          type: integer
+          description: Percentage to which the light should be dimmed to.
+          minimum: 0
+          maximum: 100
+        sentAt:
+          $ref: '#/components/schemas/sentAt'
+    sentAt:
+      type: string
+      format: date-time
+      description: Date and time when the message was sent.
+  securitySchemes:
+    saslScram:
+      type: scramSha256
+      description: Provide your username and password for SASL/SCRAM authentication
+  parameters:
+    streetlightId:
+      description: The ID of the streetlight.
+  messageTraits:
+    commonHeaders:
+      headers:
+        type: object
+        properties:
+          my-app-header:
+            type: integer
+            minimum: 0
+            maximum: 100
+  operationTraits:
+    kafka:
+      bindings:
+        kafka:
+          clientId:
+            type: string
+            enum:
+              - my-app-id

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json

@@ -55,6 +55,16 @@
                                     "authorization": "$defs/options/binding/http/authorization"
                                 },
                                 "additionalProperties": false
+                            },
+                            "kafka":
+                            {
+                                "title": "Kafka",
+                                "type": "object",
+                                "properties":
+                                {
+                                    "sasl": "$defs/options/binding/kafka/sasl"
+                                },
+                                "additionalProperties": false
                             }
                         },
                         "additionalProperties": false

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/streams/asyncapi/kafka/produce.message/client.rpt

@@ -0,0 +1,47 @@
+#
+# Copyright 2021-2023 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+property deltaMillis 0L
+property newTimestamp ${kafka:timestamp() + deltaMillis}
+
+connect "zilla://streams/asyncapi0"
+        option zilla:window 8192
+        option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${asyncapi:beginEx()
+                             .typeId(zilla:id("asyncapi"))
+                             .extension(kafka:beginEx()
+                                           .typeId(zilla:id("kafka"))
+                                           .merged()
+                                               .capabilities("PRODUCE_ONLY")
+                                               .topic("test")
+                                               .ackMode("LEADER_ONLY")
+                                               .build()
+                                            .build())
+                             .build()}
+
+connected
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .partition(0, 1)
+                                  .build()
+                              .build()}
+write "Hello, world"
+write flush