WIP mqtt

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/MqttEventContext.java

@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+package io.aklivity.zilla.runtime.binding.mqtt.internal;
+
+import java.nio.ByteBuffer;
+import java.util.function.Consumer;
+
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+
+import io.aklivity.zilla.runtime.binding.mqtt.internal.types.event.Level;
+import io.aklivity.zilla.runtime.binding.mqtt.internal.types.event.MqttEventFW;
+import io.aklivity.zilla.runtime.binding.mqtt.internal.types.event.Result;
+import io.aklivity.zilla.runtime.engine.EngineContext;
+
+public class MqttEventContext
+{
+    private static final int EVENT_BUFFER_CAPACITY = 1024;
+
+    private final MqttEventFW.Builder mqttEventRW = new MqttEventFW.Builder();
+    private final MutableDirectBuffer eventBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final Consumer<MqttEventFW> logEvent;
+
+    public MqttEventContext(
+        EngineContext context)
+    {
+        this.logEvent = context::logEvent;
+    }
+
+    public void authorization(
+        Result result,
+        Level level,
+        long originId,
+        long routedId,
+        long initialId,
+        long replyId,
+        long traceId,
+        String identity)
+    {
+        MqttEventFW event = mqttEventRW
+            .wrap(eventBuffer, 0, eventBuffer.capacity())
+            .authorization(e -> e
+                .result(r -> r.set(result))
+                .level(l -> l.set(level))
+                .originId(originId)
+                .routedId(routedId)
+                .initialId(initialId)
+                .replyId(replyId)
+                .traceId(traceId)
+                .identity(identity)
+            )
+            .build();
+        logEvent.accept(event);
+    }
+}

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java

@@ -112,6 +112,7 @@
 import io.aklivity.zilla.runtime.binding.mqtt.config.MqttPatternConfig.MqttConnectProperty;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.MqttBinding;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.MqttConfiguration;
+import io.aklivity.zilla.runtime.binding.mqtt.internal.MqttEventContext;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.MqttValidator;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.config.MqttBindingConfig;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.config.MqttRouteConfig;
@@ -167,6 +168,8 @@
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.codec.MqttUserPropertyFW;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.codec.MqttWillV4FW;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.codec.MqttWillV5FW;
+import io.aklivity.zilla.runtime.binding.mqtt.internal.types.event.Level;
+import io.aklivity.zilla.runtime.binding.mqtt.internal.types.event.Result;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.stream.AbortFW;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.stream.BeginFW;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.types.stream.DataFW;
@@ -403,6 +406,7 @@ public final class MqttServerFactory implements MqttStreamFactory
     private final Map<MqttPacketType, MqttServerDecoder> decodersByPacketTypeV5;
     private final IntSupplier supplySubscriptionId;
     private final EngineContext context;
+    private final MqttEventContext event;
 
     private int maximumPacketSize = Integer.MAX_VALUE;
 
@@ -527,6 +531,7 @@ public MqttServerFactory(
         this.decodePacketTypeByVersion.put(MQTT_PROTOCOL_VERSION_4, this::decodePacketTypeV4);
         this.decodePacketTypeByVersion.put(MQTT_PROTOCOL_VERSION_5, this::decodePacketTypeV5);
         this.unreleasedPacketIdsByClientId = unreleasedPacketIdsByClientId;
+        this.event = new MqttEventContext(context);
     }
 
     @Override
@@ -2923,6 +2928,11 @@ else if (this.authField.equals(MqttConnectProperty.PASSWORD))
                     if (credentialsMatch != null)
                     {
                         sessionAuth = guard.reauthorize(initialId, credentialsMatch);
+                        event.authorization(
+                            sessionAuth == 0 ? Result.FAILURE : Result.SUCCESS,
+                            sessionAuth == 0 ? Level.WARNING : Level.INFO,
+                            originId, routedId, initialId, replyId, traceId,
+                            guard.identity(sessionId));
                     }
                 }
 

specs/binding-mqtt.spec/src/main/resources/META-INF/zilla/mqtt.idl

@@ -257,4 +257,22 @@ scope mqtt
             int16[length] packetIds;
         }
     }
+
+    scope event
+    {
+        enum MqttEventType (uint8)
+        {
+            AUTHORIZATION (1)
+        }
+
+        struct MqttAuthorizationEvent extends core::event::BindingEvent
+        {
+            string8 identity;
+        }
+
+        union MqttEvent switch (MqttEventType)
+        {
+            case AUTHORIZATION: MqttAuthorizationEvent authorization;
+        }
+    }
 }