feedbacks

Author: bmaidics

incubator/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/AyncapiKafkaProtocol.java

@@ -14,23 +14,28 @@
  */
 package io.aklivity.zilla.runtime.binding.asyncapi.internal;
 
+import java.net.URI;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiOptionsConfig;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.Asyncapi;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiMessage;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiOperation;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiChannelView;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiMessageView;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiServerView;
-import io.aklivity.zilla.runtime.binding.http.config.HttpRequestConfig;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaOptionsConfig;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaOptionsConfigBuilder;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaSaslConfig;
 import io.aklivity.zilla.runtime.binding.kafka.config.KafkaServerConfig;
+import io.aklivity.zilla.runtime.binding.kafka.config.KafkaTopicConfig;
+import io.aklivity.zilla.runtime.binding.kafka.config.KafkaTopicConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.BindingConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.CatalogedConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfigBuilder;
-
-import java.net.URI;
-import java.util.stream.Collectors;
+import io.aklivity.zilla.runtime.model.json.config.JsonModelConfig;
 
 public class AyncapiKafkaProtocol extends AsyncapiProtocol
 {
@@ -82,7 +87,7 @@ public <C> BindingConfigBuilder<C> injectProtocolClientOptions(
                     .password(sasl.password)
                     .build()
                 .inject(this::injectKafkaServerOptions)
-                .inject(this::injectKafkaBootstrapOptions)
+                //.inject(this::injectKafkaBootstrapOptions)
                 .inject(this::injectKafkaTopicOptions)
                 .build();
     }
@@ -124,24 +129,56 @@ private <C> KafkaOptionsConfigBuilder<C> injectKafkaTopicOptions(
         {
             AsyncapiOperation operation = asyncApi.operations.get(name);
             AsyncapiChannelView channel = AsyncapiChannelView.of(asyncApi.channels, operation.channel);
-            String topic = channel.address().replaceAll("\\{[^}]+\\}", "*");
-
+            String topic = channel.address();
 
-
-
-            HttpRequestConfig.Method method = HttpRequestConfig.Method.valueOf(operation.bindings.get("http").method);
             if (channel.messages() != null && !channel.messages().isEmpty() ||
                 channel.parameters() != null && !channel.parameters().isEmpty())
             {
                 options
-                    .request()
-                    .path(path)
-                    .method(method)
-                    .inject(request -> injectContent(request, channel.messages()))
-                    .inject(request -> injectPathParams(request, channel.parameters()))
+                    .topic(KafkaTopicConfig::builder)
+                        .name(topic)
+                        .inject(topicConfig -> injectValue(topicConfig, channel.messages()))
+                        .build()
                     .build();
             }
         }
         return options;
     }
+
+    private <C> KafkaTopicConfigBuilder<C> injectValue(
+        KafkaTopicConfigBuilder<C> topic,
+        Map<String, AsyncapiMessage> messages)
+    {
+        if (messages != null)
+        {
+            if (hasJsonContentType())
+            {
+                topic
+                    .value(JsonModelConfig::builder)
+                        .catalog()
+                        .name(INLINE_CATALOG_NAME)
+                        .inject(catalog -> injectSchemas(catalog, messages))
+                        .build()
+                    .build();
+            }
+        }
+        return topic;
+    }
+
+    private <C> CatalogedConfigBuilder<C> injectSchemas(
+        CatalogedConfigBuilder<C> catalog,
+        Map<String, AsyncapiMessage> messages)
+    {
+        for (String name : messages.keySet())
+        {
+            AsyncapiMessageView message = AsyncapiMessageView.of(asyncApi.components.messages, messages.get(name));
+            String subject = message.refKey() != null ? message.refKey() : name;
+            catalog
+                .schema()
+                    .subject(subject)
+                    .build()
+                .build();
+        }
+        return catalog;
+    }
 }

incubator/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/view/AsyncapiServerView.java

@@ -14,17 +14,18 @@
  */
 package io.aklivity.zilla.runtime.binding.asyncapi.internal.view;
 
+import static org.agrona.LangUtil.rethrowUnchecked;
+
 import java.net.URI;
 import java.util.List;
 import java.util.Map;
 
-import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiServer;
-import static org.agrona.LangUtil.rethrowUnchecked;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiServer;
+
 public final class AsyncapiServerView
 {
     private final AsyncapiServer server;
@@ -40,7 +41,8 @@ public List<Map<String, List<String>>> security()
         List<Map<String, List<String>>> security = null;
         try
         {
-            security = objectMapper.readValue(server.security.toString(), new TypeReference<>() { });
+            security = objectMapper.readValue(server.security.toString(), new TypeReference<>()
+            { });
         }
         catch (JsonProcessingException e)
         {