Support obtaining protobuf schemas from schema registry for grpc services

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcBindingConfig.java

@@ -16,6 +16,7 @@
 
 import static io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.GrpcType.BASE64;
 import static io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.GrpcType.TEXT;
+import static io.aklivity.zilla.runtime.engine.catalog.CatalogHandler.NO_SCHEMA_ID;
 import static java.util.Arrays.asList;
 import static java.util.stream.Collectors.toList;
 
@@ -28,24 +29,31 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.LongFunction;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Stream;
 
 import org.agrona.AsciiSequenceView;
 import org.agrona.DirectBuffer;
 import org.agrona.MutableDirectBuffer;
+import org.agrona.collections.ObjectHashSet;
 
 import io.aklivity.zilla.runtime.binding.grpc.config.GrpcMethodConfig;
 import io.aklivity.zilla.runtime.binding.grpc.config.GrpcOptionsConfig;
+import io.aklivity.zilla.runtime.binding.grpc.config.GrpcProtobufConfig;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.Array32FW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.HttpHeaderFW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.String8FW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.GrpcMetadataFW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.GrpcType;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.HttpBeginExFW;
+import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.CatalogedConfig;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
+import io.aklivity.zilla.runtime.engine.config.SchemaConfig;
 
 public final class GrpcBindingConfig
 {
@@ -56,28 +64,41 @@ public final class GrpcBindingConfig
     private static final byte[] HEADER_BIN_SUFFIX = new byte[4];
     private static final byte[] GRPC_PREFIX = "grpc-".getBytes();
     private static final byte[] BIN_SUFFIX = "-bin".getBytes();
-    private final HttpGrpcHeaderHelper helper;
 
     public final long id;
     public final String name;
     public final KindConfig kind;
     public final GrpcOptionsConfig options;
     public final List<GrpcRouteConfig> routes;
 
+    private final GrpcProtobufParser parser;
+    private final HttpGrpcHeaderHelper helper;
+    private final Set<GrpcCatalogSchema> catalogs;
 
     public GrpcBindingConfig(
         BindingConfig binding,
-        MutableDirectBuffer metadataBuffer)
+        MutableDirectBuffer metadataBuffer,
+        LongFunction<CatalogHandler> supplyCatalog)
     {
         this.id = binding.id;
         this.name = binding.name;
         this.kind = binding.kind;
         this.options = GrpcOptionsConfig.class.cast(binding.options);
         this.routes = binding.routes.stream().map(GrpcRouteConfig::new).collect(toList());
+        this.parser = new GrpcProtobufParser();
         this.helper = new HttpGrpcHeaderHelper(metadataBuffer);
+        Set<GrpcCatalogSchema> catalogs = new ObjectHashSet<>();
+        for (CatalogedConfig catalog : binding.catalogs)
+        {
+            CatalogHandler handler = supplyCatalog.apply(catalog.id);
+            for (SchemaConfig schema : catalog.schemas)
+            {
+                catalogs.add(new GrpcCatalogSchema(handler, schema.subject, schema.version));
+            }
+        }
+        this.catalogs = catalogs;
     }
 
-
     public GrpcRouteConfig resolve(
         long authorization,
         CharSequence service,
@@ -107,13 +128,12 @@ public GrpcMethodResult resolveMethod(
             final CharSequence serviceName = serviceNameHeader != null ? serviceNameHeader : matcher.group(SERVICE_NAME);
             final String methodName = matcher.group(METHOD);
 
-            final GrpcMethodConfig method = options.protobufs.stream()
-                .map(p -> p.services.stream().filter(s -> s.service.equals(serviceName)).findFirst().orElse(null))
-                .filter(Objects::nonNull)
-                .map(s -> s.methods.stream().filter(m -> m.method.equals(methodName)).findFirst().orElse(null))
-                .filter(Objects::nonNull)
-                .findFirst()
-                .orElse(null);
+            GrpcMethodConfig method = resolveMethod(catalogs, serviceName, methodName);
+
+            if (method == null && options != null)
+            {
+                method = resolveMethod(options.protobufs, serviceName, methodName);
+            }
 
             if (method != null)
             {
@@ -133,6 +153,36 @@ public GrpcMethodResult resolveMethod(
         return methodResolver;
     }
 
+    private GrpcMethodConfig resolveMethod(
+        Set<GrpcCatalogSchema> catalogs,
+        CharSequence serviceName,
+        String methodName)
+    {
+        return resolveMethod(catalogs.stream().map(GrpcCatalogSchema::resolveProtobuf), serviceName, methodName);
+    }
+
+    private GrpcMethodConfig resolveMethod(
+        List<GrpcProtobufConfig> protobufs,
+        CharSequence serviceName,
+        String methodName)
+    {
+        return resolveMethod(protobufs.stream(), serviceName, methodName);
+    }
+
+    private GrpcMethodConfig resolveMethod(
+        Stream<GrpcProtobufConfig> protobufs,
+        CharSequence serviceName,
+        String methodName)
+    {
+        return protobufs
+            .map(p -> p.services.stream().filter(s -> s.service.equals(serviceName)).findFirst().orElse(null))
+            .filter(Objects::nonNull)
+            .map(s -> s.methods.stream().filter(m -> m.method.equals(methodName)).findFirst().orElse(null))
+            .filter(Objects::nonNull)
+            .findFirst()
+            .orElse(null);
+    }
+
     private static final class HttpGrpcHeaderHelper
     {
         private static final Pattern PERIOD_PATTERN = Pattern.compile("([0-9]+)([HMSmun])");
@@ -187,7 +237,6 @@ private static final class HttpGrpcHeaderHelper
         public String16FW authority;
         public String16FW te;
 
-
         HttpGrpcHeaderHelper(
             MutableDirectBuffer metadataBuffer)
         {
@@ -350,4 +399,39 @@ private long parsePeriod(
             return milliseconds;
         }
     }
+
+    final class GrpcCatalogSchema
+    {
+        final CatalogHandler handler;
+        final String subject;
+        final String version;
+
+        GrpcProtobufConfig protobuf;
+
+        int schemaId = NO_SCHEMA_ID;
+
+        GrpcCatalogSchema(
+            CatalogHandler handler,
+            String subject,
+            String version)
+        {
+            this.handler = handler;
+            this.subject = subject;
+            this.version = version;
+        }
+
+        private GrpcProtobufConfig resolveProtobuf()
+        {
+            final int newSchemaId = handler.resolve(subject, version);
+
+            if (schemaId != newSchemaId)
+            {
+                schemaId = newSchemaId;
+                String schema = handler.resolve(schemaId);
+                protobuf = parser.parse(null, schema);
+            }
+
+            return protobuf;
+        }
+    }
 }

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcOptionsConfigAdapter.java

@@ -17,7 +17,6 @@
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
-import java.util.Set;
 import java.util.function.Function;
 
 import jakarta.json.Json;
@@ -29,26 +28,19 @@
 import jakarta.json.JsonValue;
 import jakarta.json.bind.adapter.JsonbAdapter;
 
-import org.agrona.collections.ObjectHashSet;
-import org.antlr.v4.runtime.BailErrorStrategy;
-import org.antlr.v4.runtime.CharStream;
-import org.antlr.v4.runtime.CharStreams;
-import org.antlr.v4.runtime.CommonTokenStream;
-import org.antlr.v4.runtime.tree.ParseTreeWalker;
-
 import io.aklivity.zilla.runtime.binding.grpc.config.GrpcOptionsConfig;
 import io.aklivity.zilla.runtime.binding.grpc.config.GrpcProtobufConfig;
-import io.aklivity.zilla.runtime.binding.grpc.config.GrpcServiceConfig;
 import io.aklivity.zilla.runtime.binding.grpc.internal.GrpcBinding;
-import io.aklivity.zilla.runtime.binding.grpc.internal.parser.Protobuf3Lexer;
-import io.aklivity.zilla.runtime.binding.grpc.internal.parser.Protobuf3Parser;
 import io.aklivity.zilla.runtime.engine.config.ConfigAdapterContext;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfigAdapterSpi;
 
 public final class GrpcOptionsConfigAdapter implements OptionsConfigAdapterSpi, JsonbAdapter<OptionsConfig, JsonObject>
 {
     private static final String SERVICES_NAME = "services";
+
+    private final GrpcProtobufParser parser = new GrpcProtobufParser();
+
     private Function<String, String> readURL;
 
     @Override
@@ -111,18 +103,8 @@ private GrpcProtobufConfig asProtobuf(
         JsonValue value)
     {
         final String location = ((JsonString) value).getString();
-        final String protoService = readURL.apply(location);
-        CharStream input = CharStreams.fromString(protoService);
-        Protobuf3Lexer lexer = new Protobuf3Lexer(input);
-        CommonTokenStream tokens = new CommonTokenStream(lexer);
-
-        Protobuf3Parser parser = new Protobuf3Parser(tokens);
-        parser.setErrorHandler(new BailErrorStrategy());
-        ParseTreeWalker walker = new ParseTreeWalker();
-        Set<GrpcServiceConfig> services = new ObjectHashSet<>();
-        GrpcServiceDefinitionListener listener = new GrpcServiceDefinitionListener(services);
-        walker.walk(listener, parser.proto());
-
-        return new GrpcProtobufConfig(location, services);
+        final String protobuf = readURL.apply(location);
+
+        return parser.parse(location, protobuf);
     }
 }

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcProtobufParser.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2021-2023 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.grpc.internal.config;
+
+import java.util.Set;
+
+import org.agrona.collections.ObjectHashSet;
+import org.antlr.v4.runtime.BailErrorStrategy;
+import org.antlr.v4.runtime.CharStream;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.tree.ParseTreeWalker;
+
+import io.aklivity.zilla.runtime.binding.grpc.config.GrpcProtobufConfig;
+import io.aklivity.zilla.runtime.binding.grpc.config.GrpcServiceConfig;
+import io.aklivity.zilla.runtime.binding.grpc.internal.parser.Protobuf3Lexer;
+import io.aklivity.zilla.runtime.binding.grpc.internal.parser.Protobuf3Parser;
+
+public final class GrpcProtobufParser
+{
+    private final ParseTreeWalker walker;
+    private final BailErrorStrategy errorStrategy;
+    private final Protobuf3Lexer lexer;
+    private CommonTokenStream tokens;
+    private final Protobuf3Parser parser;
+
+    public GrpcProtobufParser()
+    {
+        this.walker = new ParseTreeWalker();
+        this.errorStrategy = new BailErrorStrategy();
+        this.lexer = new Protobuf3Lexer(null);
+        this.parser = new Protobuf3Parser(null);
+        this.tokens = new CommonTokenStream(lexer);
+        parser.setErrorHandler(errorStrategy);
+    }
+
+    public GrpcProtobufConfig parse(
+        String location,
+        String schema)
+    {
+        CharStream input = CharStreams.fromString(schema);
+        lexer.reset();
+        lexer.setInputStream(input);
+
+        tokens.setTokenSource(lexer);
+        parser.setTokenStream(tokens);
+
+        Set<GrpcServiceConfig> services = new ObjectHashSet<>();
+        GrpcServiceDefinitionListener listener = new GrpcServiceDefinitionListener(services);
+        walker.walk(listener, parser.proto());
+
+        return new GrpcProtobufConfig(location, services);
+    }
+}

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/stream/GrpcClientFactory.java

@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.grpc.internal.stream;
 
+import java.util.function.LongFunction;
 import java.util.function.LongUnaryOperator;
 
 import org.agrona.DirectBuffer;
@@ -50,6 +51,7 @@
 import io.aklivity.zilla.runtime.engine.EngineContext;
 import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
+import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 
 public class GrpcClientFactory implements GrpcStreamFactory
@@ -114,6 +116,7 @@ public class GrpcClientFactory implements GrpcStreamFactory
     private final MutableDirectBuffer metadataBuffer;
     private final MutableDirectBuffer extBuffer;
     private final BindingHandler streamFactory;
+    private final LongFunction<CatalogHandler> supplyCatalog;
     private final LongUnaryOperator supplyInitialId;
     private final LongUnaryOperator supplyReplyId;
     private final int httpTypeId;
@@ -130,6 +133,7 @@ public GrpcClientFactory(
         this.metadataBuffer = new UnsafeBuffer(new byte[writeBuffer.capacity()]);
         this.extBuffer = new UnsafeBuffer(new byte[writeBuffer.capacity()]);
         this.streamFactory = context.streamFactory();
+        this.supplyCatalog = context::supplyCatalog;
         this.supplyInitialId = context::supplyInitialId;
         this.supplyReplyId = context::supplyReplyId;
         this.httpTypeId = context.supplyTypeId(HTTP_TYPE_NAME);
@@ -159,7 +163,7 @@ public int routedTypeId()
     public void attach(
         BindingConfig binding)
     {
-        GrpcBindingConfig grpcBinding = new GrpcBindingConfig(binding, metadataBuffer);
+        GrpcBindingConfig grpcBinding = new GrpcBindingConfig(binding, metadataBuffer, supplyCatalog);
         bindings.put(binding.id, grpcBinding);
     }
 

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/stream/GrpcServerFactory.java

@@ -23,6 +23,7 @@
 import static java.time.Instant.now;
 
 import java.util.function.Consumer;
+import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
 
@@ -62,6 +63,7 @@
 import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
+import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.concurrent.Signaler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 
@@ -134,6 +136,7 @@ public final class GrpcServerFactory implements GrpcStreamFactory
     private final BufferPool bufferPool;
     private final Signaler signaler;
     private final BindingHandler streamFactory;
+    private final LongFunction<CatalogHandler> supplyCatalog;
     private final LongUnaryOperator supplyInitialId;
     private final LongUnaryOperator supplyReplyId;
     private final LongSupplier supplyTraceId;
@@ -235,6 +238,7 @@ public GrpcServerFactory(
         this.bufferPool = context.bufferPool();
         this.signaler = context.signaler();
         this.streamFactory = context.streamFactory();
+        this.supplyCatalog = context::supplyCatalog;
         this.supplyInitialId = context::supplyInitialId;
         this.supplyReplyId = context::supplyReplyId;
         this.supplyTraceId = context::supplyTraceId;
@@ -259,7 +263,7 @@ public int routedTypeId()
     public void attach(
         BindingConfig binding)
     {
-        GrpcBindingConfig grpcBinding = new GrpcBindingConfig(binding, metadataBuffer);
+        GrpcBindingConfig grpcBinding = new GrpcBindingConfig(binding, metadataBuffer, supplyCatalog);
         bindings.put(binding.id, grpcBinding);
     }