Is it possible to pass the byte[] of the proto object to the Response Observer onNext call instead of the deserialized object?

Basically, I have a grpc service that on a request goes and pulls some proto objects from a database instance, and sends them to a client.
In the database, the objects are actually just stored as a byte[], which comes from the proto objects .toByteArray() call.

Right now I pull the bytes from the database, deserialize them and then send them to the response observer via onNext(). I know the onNext call just serializes them again to send them off.

I was wondering it's possible for me to avoid this duplicate effort of serializing/deserializing and just give the byte[] directly to the response observer?

With the generated code, no, because of type safety. On client-side it is "pretty easy" to avoid the generate code and to use ClientCalls directly for one RPC. You modify the MethodDescriptor to replace the Marshaller. That'd look something like:

class MyByteArrayMarshaller implements MethodDescriptor.Marshaller<byte[]> {
  @Override byte[] parse(InputStream stream) { ByteStreams.toByteArray(stream); }
  @Override InputStream stream(byte[] b) { return new ByteArrayInputStream(b); }
}

MethodDescriptor<Foo, Bar> desc = FooGrpc.getBarMethod();
MethodDescriptor<byte[], Bar> myDesc = desc
    .toBuilder(new MyByteArrayMarshaller(), desc.getResponseMarshaller()).build();

// And now do an RPC when you want. This uses the API used by stubs
Bar resp = ClientCalls.blockingUnaryCall(channel, myDesc, CallOptions.DEFAULT, bytes);

Server-side is more annoying. On server-side you want to modify ServerServiceDefinition, to inject your own callback. That object is normally returned by FooGrpc.bindService().

// Let's say you only want this optimization for one method, but there's other
// methods on the service
class MyFooImpl extends FooImplBase {
  // Normally this would have been StreamObserver<Bar>
  public void bar(Foo request, StreamObserver<byte[]> responseObs) {
    ...
  }
  ...
}

MyFooImpl myFooImpl = new MyFooImpl();
ServerServiceDefinition def = myFooImpl.bindService();
ServerServiceDefinition.Builder b = ServerServiceDefintion.builder(def.getServiceDescriptor());
for (ServerMethodDefinition<?,?> smd : def.getMethods()) {
  // Skip the one method we want to replace
  if ("bar".equals(smd.getMethodDescriptor().getBareMethodName()) {
    continue;
  }
  b.addMethod(smd);
}
// myDesc, as done on client-side except swap out the response instead of request
b.addMethod(myDesc, ServerCalls.asyncUnaryCall(myFooImpl::bar));
ServerServiceDefinition myDef = b.build();

// And now you could add that to a server
serverBuilder.addService(myDef);

So very possible, but clearly annoying. There is type-safety though! Everything else works like normal. Interceptors are fine. You're using the same StreamObserver implementation/handling as normal Stubs. So the main cost is the boilerplate.

Ok, deleted a few previous comments, because I figured out what I was doing wrong; Ultimately, when I run this, I get a "Bound method for package.service.Service/RequestRPC not same instance as method in service descriptor" error message

Ugh, I forgot about the parallel entries in the ServiceDescriptor (def.getServiceDescriptor()). I think that can be worked around easily with ServerServiceDefintion.builder(def.getServiceDescriptor().getName()) instead of ServerServiceDefintion.builder(def.getServiceDescriptor()). It would work, but lose reflection information. For reflection, it'd require the same sort of pattern I showed, but for ServiceDescriptor in addition.

(Even before this ServiceDescriptor issue) I do think we need some APIs here to make this more convenient.

(Edit: Earlier I thought that serverServiceDefinition.toBuilder() and replaceMethod() would be sufficient. But that wouldn't be enough with ServiceDescriptor causing trouble as well.)

Ok, well I really appreciate you looking into this. Purely anecdotal but here is a screenshot from some of the profiling I did while looking at this possibility. The byte[] is the amount of memory used just pulling the bytes from the database. The Builder is the memory allocated for the serializing and deserializing of the said object. Definitely would increase efficiency if it became possible at some point in the future to skip the serializing and just send the bytes.

Thanks again!

I wasn't saying it wasn't possible. Just more annoying. Here's a full working example for the hello world service, with both the client and server-side changes as examples (you don't need both; it's just easier to understand what is going on on server-side after understanding client-side).

diff --git a/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java b/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java
index 6b186facf..870438d0e 100644
--- a/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java
+++ b/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java
@@ -16,11 +16,18 @@
 
 package io.grpc.examples.helloworld;
 
+import com.google.common.io.ByteStreams;
+import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.Grpc;
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
+import io.grpc.MethodDescriptor;
 import io.grpc.StatusRuntimeException;
+import io.grpc.stub.ClientCalls;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -31,7 +38,7 @@ import java.util.logging.Logger;
 public class HelloWorldClient {
   private static final Logger logger = Logger.getLogger(HelloWorldClient.class.getName());
 
-  private final GreeterGrpc.GreeterBlockingStub blockingStub;
+  private final Channel channel;
 
   /** Construct client for accessing HelloWorld server using the existing channel. */
   public HelloWorldClient(Channel channel) {
@@ -39,7 +46,7 @@ public class HelloWorldClient {
     // shut it down.
 
     // Passing Channels to code makes code easier to test and makes it easier to reuse Channels.
-    blockingStub = GreeterGrpc.newBlockingStub(channel);
+    this.channel = channel;
   }
 
   /** Say hello to server. */
@@ -47,8 +54,12 @@ public class HelloWorldClient {
     logger.info("Will try to greet " + name + " ...");
     HelloRequest request = HelloRequest.newBuilder().setName(name).build();
     HelloReply response;
+    MethodDescriptor<HelloRequest, HelloReply> desc = GreeterGrpc.getSayHelloMethod();
+    MethodDescriptor<byte[], HelloReply> myDesc
+        = desc.toBuilder(new MyByteArrayMarshaller(), desc.getResponseMarshaller()).build();
     try {
-      response = blockingStub.sayHello(request);
+      response = ClientCalls.blockingUnaryCall(
+          channel, myDesc, CallOptions.DEFAULT, request.toByteArray());
     } catch (StatusRuntimeException e) {
       logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
       return;
@@ -97,4 +108,18 @@ public class HelloWorldClient {
       channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
     }
   }
+
+  public static final class MyByteArrayMarshaller implements MethodDescriptor.Marshaller<byte[]> {
+    @Override public byte[] parse(InputStream stream) {
+      try {
+        return ByteStreams.toByteArray(stream);
+      } catch (IOException ex) {
+        throw new RuntimeException(ex);
+      }
+    }
+
+    @Override public InputStream stream(byte[] b) {
+      return new ByteArrayInputStream(b);
+    }
+  }
 }
diff --git a/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java b/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java
index 810275870..b1ce9eb53 100644
--- a/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java
+++ b/examples/src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java
@@ -18,7 +18,13 @@ package io.grpc.examples.helloworld;
 
 import io.grpc.Grpc;
 import io.grpc.InsecureServerCredentials;
+import io.grpc.MethodDescriptor;
 import io.grpc.Server;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerMethodDefinition;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.ServiceDescriptor;
+import io.grpc.stub.ServerCalls;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -36,7 +42,7 @@ public class HelloWorldServer {
     /* The port on which the server should run */
     int port = 50051;
     server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
-        .addService(new GreeterImpl())
+        .addService(new GreeterImpl().myBindService())
         .build()
         .start();
     logger.info("Server started, listening on " + port);
@@ -81,11 +87,42 @@ public class HelloWorldServer {
 
   static class GreeterImpl extends GreeterGrpc.GreeterImplBase {
 
-    @Override
-    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
+    public void mySayHello(HelloRequest req, StreamObserver<byte[]> responseObserver) {
       HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
-      responseObserver.onNext(reply);
+      responseObserver.onNext(reply.toByteArray());
       responseObserver.onCompleted();
     }
+
+    public ServerServiceDefinition myBindService() {
+      MethodDescriptor<HelloRequest, HelloReply> desc = GreeterGrpc.getSayHelloMethod();
+      MethodDescriptor<HelloRequest, byte[]> myDesc = desc.toBuilder(
+          desc.getRequestMarshaller(), new HelloWorldClient.MyByteArrayMarshaller()).build();
+      return replaceMethod(bindService(), myDesc, ServerCalls.asyncUnaryCall(this::mySayHello));
+    }
+
+    private static <ReqT, RespT> ServerServiceDefinition replaceMethod(
+        ServerServiceDefinition def,
+        MethodDescriptor<ReqT, RespT> desc,
+        ServerCallHandler<ReqT, RespT> handler) {
+      ServiceDescriptor srvDesc = def.getServiceDescriptor();
+      ServiceDescriptor.Builder srvBuilder = ServiceDescriptor.newBuilder(srvDesc.getName())
+          .setSchemaDescriptor(srvDesc.getSchemaDescriptor())
+          .addMethod(desc);
+      for (MethodDescriptor<?,?> md : srvDesc.getMethods()) {
+        if (desc.getFullMethodName().equals(md.getFullMethodName())) {
+          continue;
+        }
+        srvBuilder.addMethod(md);
+      }
+      ServerServiceDefinition.Builder b = ServerServiceDefinition.builder(srvBuilder.build())
+          .addMethod(desc, handler);
+      for (ServerMethodDefinition<?,?> smd : def.getMethods()) {
+        if (desc.getFullMethodName().equals(smd.getMethodDescriptor().getFullMethodName())) {
+          continue;
+        }
+        b.addMethod(smd);
+      }
+      return b.build();
+    }
   }
 }

@JonSobocinski

@JonSobocinski, Were you able to progress with the provided example?

Added an example for this approach in #10112

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Repository navigation

Is it possible to pass the byte[] of the proto object to the Response Observer onNext call instead of the deserialized object? #9707

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Sunbelt Computer Software

PL/B Language Development and Support

Is it possible to pass the byte[] of the proto object to the Response Observer onNext call instead of the deserialized object? #9707

Description

Activity

ejona86 commented on Nov 22, 2022

JonSobocinski commented on Dec 8, 2022

ejona86 commented on Dec 8, 2022

JonSobocinski commented on Dec 12, 2022

ejona86 commented on Jan 3, 2023

larry-safran commented on Jan 24, 2023

ejona86 commented on Aug 15, 2023

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions