record unresolved $ref in OpenAPI

examples/asyncapi.mqtt.kafka.proxy/.github/test.sh

@@ -24,7 +24,7 @@ for i in $(seq 1 5); do
     break
   fi
 
-  sleep 2
+  sleep 5
 done
 
 OUTPUT=$(

examples/asyncapi.mqtt.proxy/.github/test.sh

@@ -24,7 +24,7 @@ for i in $(seq 1 5); do
     break
   fi
 
-  sleep 2
+  sleep 5
 done
 
 OUTPUT=$(

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/config/composite/OpenapiCompositeGenerator.java

@@ -19,9 +19,12 @@
 import java.io.StringReader;
 import java.io.StringWriter;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -67,6 +70,8 @@
 
 public abstract class OpenapiCompositeGenerator
 {
+    private final Set<String> unresolved = new LinkedHashSet<>();
+
     public final OpenapiCompositeConfig generate(
         OpenapiBindingConfig binding)
     {
@@ -88,15 +93,22 @@ public final OpenapiCompositeConfig generate(
                     specification.servers == null || specification.servers.isEmpty()
                         ? List.of(OpenapiServerConfig.builder().build())
                         : specification.servers;
-                final OpenapiView asyncapi = OpenapiView.of(tagIndex++, label, parser.parse(payload), configs);
+                final OpenapiView openapi = OpenapiView.of(tagIndex++, label, parser.parse(payload), configs);
+
+                unresolved.addAll(openapi.unresolvedRefs());
 
-                schemas.add(new OpenapiSchemaConfig(label, schemaId, asyncapi));
+                schemas.add(new OpenapiSchemaConfig(label, schemaId, openapi));
             }
         }
 
         return generate(binding, schemas);
     }
 
+    public final Collection<String> unresolvedRefs()
+    {
+        return unresolved;
+    }
+
     protected abstract OpenapiCompositeConfig generate(
         OpenapiBindingConfig binding,
         List<OpenapiSchemaConfig> schemas);

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/event/OpenapiEventContext.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES 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.openapi.internal.event;
+
+import static io.aklivity.zilla.runtime.binding.openapi.internal.types.event.OpenapiEventType.UNRESOLVED_REF;
+
+import java.nio.ByteBuffer;
+import java.time.Clock;
+
+import org.agrona.concurrent.AtomicBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+
+import io.aklivity.zilla.runtime.binding.openapi.internal.OpenapiBinding;
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.event.EventFW;
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.event.OpenapiEventExFW;
+import io.aklivity.zilla.runtime.engine.EngineContext;
+import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
+
+public class OpenapiEventContext
+{
+    private static final int EVENT_BUFFER_CAPACITY = 1024;
+
+    private final AtomicBuffer eventBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final AtomicBuffer extensionBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final EventFW.Builder eventRW = new EventFW.Builder();
+    private final OpenapiEventExFW.Builder openapiEventExRW = new OpenapiEventExFW.Builder();
+    private final int openapiTypeId;
+    private final int unresolvedRef;
+    private final MessageConsumer eventWriter;
+    private final Clock clock;
+
+    public OpenapiEventContext(
+        EngineContext context)
+    {
+        this.openapiTypeId = context.supplyTypeId(OpenapiBinding.NAME);
+        this.unresolvedRef = context.supplyEventId("binding.openapi.unresolved.ref");
+        this.eventWriter = context.supplyEventWriter();
+        this.clock = context.clock();
+    }
+
+    public void unresolvedRef(
+        long bindingId,
+        String ref)
+    {
+        OpenapiEventExFW extension = openapiEventExRW
+            .wrap(extensionBuffer, 0, extensionBuffer.capacity())
+            .unresolvedRef(e -> e
+                .typeId(UNRESOLVED_REF.value())
+                .ref(ref)
+            )
+            .build();
+        EventFW event = eventRW
+            .wrap(eventBuffer, 0, eventBuffer.capacity())
+            .id(unresolvedRef)
+            .timestamp(clock.millis())
+            .traceId(0L)
+            .namespacedId(bindingId)
+            .extension(extension.buffer(), extension.offset(), extension.limit())
+            .build();
+        eventWriter.accept(openapiTypeId, event.buffer(), event.offset(), event.limit());
+    }
+}

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/event/OpenapiEventFormatter.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES 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.openapi.internal.event;
+
+import org.agrona.DirectBuffer;
+
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.String16FW;
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.event.EventFW;
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.event.OpenapiEventExFW;
+import io.aklivity.zilla.runtime.binding.openapi.internal.types.event.OpenapiUnresolvedRefExFW;
+import io.aklivity.zilla.runtime.engine.Configuration;
+import io.aklivity.zilla.runtime.engine.event.EventFormatterSpi;
+
+public final class OpenapiEventFormatter implements EventFormatterSpi
+{
+    private final EventFW eventRO = new EventFW();
+    private final OpenapiEventExFW openapiEventExRO = new OpenapiEventExFW();
+
+    OpenapiEventFormatter(
+        Configuration config)
+    {
+    }
+
+    public String format(
+        DirectBuffer buffer,
+        int index,
+        int length)
+    {
+        final EventFW event = eventRO.wrap(buffer, index, index + length);
+        final OpenapiEventExFW extension = openapiEventExRO
+            .wrap(event.extension().buffer(), event.extension().offset(), event.extension().limit());
+        String result = null;
+        switch (extension.kind())
+        {
+        case UNRESOLVED_REF:
+        {
+            OpenapiUnresolvedRefExFW ex = extension.unresolvedRef();
+            result = String.format("Unresolved reference (%s).", asString(ex.ref()));
+            break;
+        }
+        }
+        return result;
+    }
+
+    private static String asString(
+        String16FW stringFW)
+    {
+        String s = stringFW.asString();
+        return s == null ? "" : s;
+    }
+}

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/event/OpenapiEventFormatterFactory.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES 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.openapi.internal.event;
+
+import io.aklivity.zilla.runtime.binding.openapi.internal.OpenapiBinding;
+import io.aklivity.zilla.runtime.engine.Configuration;
+import io.aklivity.zilla.runtime.engine.event.EventFormatterFactorySpi;
+
+public final class OpenapiEventFormatterFactory implements EventFormatterFactorySpi
+{
+    @Override
+    public OpenapiEventFormatter create(
+        Configuration config)
+    {
+        return new OpenapiEventFormatter(config);
+    }
+
+    @Override
+    public String type()
+    {
+        return OpenapiBinding.NAME;
+    }
+}

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/model/resolver/AbstractOpenapiResolver.java

@@ -15,6 +15,7 @@
 package io.aklivity.zilla.runtime.binding.openapi.internal.model.resolver;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -24,20 +25,28 @@ public abstract class AbstractOpenapiResolver<T extends AbstractOpenapiResolvabl
 {
     private final Map<String, T> resolvables;
     private final Matcher matcher;
+    private final Set<String> unresolved;
 
     public AbstractOpenapiResolver(
         Map<String, T> resolvables,
-        Pattern pattern)
+        Pattern pattern,
+        Set<String> unresolved)
     {
         this.resolvables = resolvables;
         this.matcher = pattern.matcher("");
+        this.unresolved = unresolved;
     }
 
     public final T resolve(
         T model)
     {
         final String key = resolveRef(model.ref);
-        return key != null ? resolvables.get(key) : model;
+        T candidate = key != null ? resolvables.getOrDefault(key, model) : model;
+        if (candidate.ref != null)
+        {
+            unresolved.add(candidate.ref);
+        }
+        return candidate;
     }
 
     public T resolve(

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/model/resolver/OpenapiHeaderResolver.java

@@ -16,6 +16,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
@@ -24,12 +25,14 @@
 public final class OpenapiHeaderResolver extends AbstractOpenapiResolver<OpenapiHeader>
 {
     public OpenapiHeaderResolver(
-        Openapi model)
+        Openapi model,
+        Set<String> unresolved)
     {
         super(
             Optional.ofNullable(model.components)
                 .map(c -> c.headers)
                 .orElseGet(Map::of),
-            Pattern.compile("#/components/headers/(.+)"));
+            Pattern.compile("#/components/headers/(.+)"),
+            unresolved);
     }
 }

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/model/resolver/OpenapiLinkResolver.java

@@ -16,6 +16,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
@@ -24,12 +25,14 @@
 public final class OpenapiLinkResolver extends AbstractOpenapiResolver<OpenapiLink>
 {
     public OpenapiLinkResolver(
-        Openapi model)
+        Openapi model,
+        Set<String> unresolved)
     {
         super(
             Optional.ofNullable(model.components)
                 .map(c -> c.links)
                 .orElseGet(Map::of),
-            Pattern.compile("#/components/links/(.+)"));
+            Pattern.compile("#/components/links/(.+)"),
+            unresolved);
     }
 }

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/model/resolver/OpenapiParameterResolver.java

@@ -16,6 +16,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
@@ -24,12 +25,14 @@
 public final class OpenapiParameterResolver extends AbstractOpenapiResolver<OpenapiParameter>
 {
     public OpenapiParameterResolver(
-        Openapi model)
+        Openapi model,
+        Set<String> unresolved)
     {
         super(
             Optional.ofNullable(model.components)
                 .map(c -> c.parameters)
                 .orElseGet(Map::of),
-            Pattern.compile("#/components/parameters/(.+)"));
+            Pattern.compile("#/components/parameters/(.+)"),
+            unresolved);
     }
 }

runtime/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/model/resolver/OpenapiRequestBodyResolver.java

@@ -16,6 +16,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
@@ -24,12 +25,14 @@
 public final class OpenapiRequestBodyResolver extends AbstractOpenapiResolver<OpenapiRequestBody>
 {
     public OpenapiRequestBodyResolver(
-        Openapi model)
+        Openapi model,
+        Set<String> unresolved)
     {
         super(
             Optional.ofNullable(model.components)
                 .map(c -> c.requestBodies)
                 .orElseGet(Map::of),
-            Pattern.compile("#/components/requestBodies/(.+)"));
+            Pattern.compile("#/components/requestBodies/(.+)"),
+            unresolved);
     }
 }