Support gRPC client stream/unary oneway (#1384)

aklivity · Feb 4, 2025 · cf16ddb · cf16ddb
1 parent 869533f
commit cf16ddb
Show file tree

Hide file tree

Showing 17 changed files with 1,041 additions and 58 deletions.
diff --git a/...y/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaWithProduceConfigAdapter.java b/...y/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaWithProduceConfigAdapter.java
@@ -101,7 +101,11 @@ public GrpcKafkaWithConfig adaptFromJson(
             }
         }
 
-        String newReplyTo = object.getString(REPLY_TO_NAME);
+        String newReplyTo = null;
+        if (object.containsKey(REPLY_TO_NAME))
+        {
+            newReplyTo = object.getString(REPLY_TO_NAME);
+        }
 
         return new GrpcKafkaWithConfig(
             new GrpcKafkaWithProduceConfig(newTopic, newProduceAcks, newProduceKey, newOverrides, newReplyTo));

diff --git a/...aklivity/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaWithProduceResult.java b/...aklivity/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaWithProduceResult.java
@@ -89,12 +89,15 @@ public class GrpcKafkaWithProduceResult
         this.nameBuffer = new ExpandableDirectByteBuffer();
         this.nameBuffer.putStringWithoutLengthAscii(0, META_PREFIX);
 
-        hash.updateHash(correlation.service.value());
-        hash.updateHash(service.value());
-        hash.updateHash(correlation.method.value());
-        hash.updateHash(method.value());
-        hash.updateHash(correlation.replyTo.value());
-        hash.updateHash(replyTo.value());
+        if (hasReplyTo())
+        {
+            hash.updateHash(correlation.service.value());
+            hash.updateHash(service.value());
+            hash.updateHash(correlation.method.value());
+            hash.updateHash(method.value());
+            hash.updateHash(correlation.replyTo.value());
+            hash.updateHash(replyTo.value());
+        }
 
         if (overrides != null)
         {
@@ -153,10 +156,14 @@ public void headers(
             overrides.forEach(o -> builder.item(o::header));
         }
 
-        builder.item(this::service);
-        builder.item(this::method);
-        builder.item(this::replyTo);
-        builder.item(this::correlationId);
+        if (hasReplyTo())
+        {
+            builder.item(this::service);
+            builder.item(this::method);
+            builder.item(this::replyTo);
+            builder.item(this::correlationId);
+        }
+
         metadata.forEach(m -> builder.item(i -> metadata(i, m)));
     }
 
@@ -241,4 +248,9 @@ public void filters(
                     .valueLen(hashCorrelationId.sizeof())
                     .value(hashCorrelationId.value(), 0, hashCorrelationId.sizeof()))));
     }
+
+    public boolean hasReplyTo()
+    {
+        return replyTo != null && replyTo.value() != null;
+    }
 }