fix(implementation): Check if data is Buffer already (#293)

* fix(implementation): Check if data is Buffer already
Sometimes we need to send bytes to downstream,
For example sending data with avro format to Kafka, we should change its format anymore.

Fixes #292

Signed-off-by: Alex Ji <[email protected]>

* fix(implementation): Add unit test and fix invoker as well

Signed-off-by: Alex Ji <[email protected]>

* data should be const

Signed-off-by: Alex Ji <[email protected]>

* Change serializeGrpc method so that it can return

both the data and the content-type based on if it's a Buffer or not.

Signed-off-by: Alex Ji <[email protected]>

* Refactor Serializer.util.ts

Signed-off-by: Alex Ji <[email protected]>

Author: James-JLC

src/implementation/Client/GRPCClient/binding.ts

@@ -14,6 +14,7 @@ limitations under the License.
 import GRPCClient from './GRPCClient';
 import { InvokeBindingRequest, InvokeBindingResponse } from '../../../proto/dapr/proto/runtime/v1/dapr_pb';
 import IClientBinding from '../../../interfaces/Client/IClientBinding';
+import * as SerializerUtil from "../../../utils/Serializer.util";
 
 // https://docs.dapr.io/reference/api/bindings_api/
 export default class GRPCClientBinding implements IClientBinding {
@@ -30,7 +31,7 @@ export default class GRPCClientBinding implements IClientBinding {
     const msgService = new InvokeBindingRequest();
     msgService.setName(bindingName);
     msgService.setOperation(operation);
-    msgService.setData(Buffer.from(JSON.stringify(data), "utf-8"));
+    msgService.setData(SerializerUtil.serializeGrpc(data).serializedData);
 
     return new Promise((resolve, reject) => {
       const client = this.client.getClient();

src/implementation/Client/GRPCClient/invoker.ts

@@ -19,6 +19,7 @@ import { HTTPExtension, InvokeRequest, InvokeResponse } from '../../../proto/dap
 import { InvokeServiceRequest } from '../../../proto/dapr/proto/runtime/v1/dapr_pb';
 import * as HttpVerbUtil from "../../../utils/HttpVerb.util";
 import IClientInvoker from '../../../interfaces/Client/IClientInvoker';
+import * as SerializerUtil from "../../../utils/Serializer.util"
 
 // https://docs.dapr.io/reference/api/service_invocation_api/
 export default class GRPCClientInvoker implements IClientInvoker {
@@ -54,13 +55,14 @@ export default class GRPCClientInvoker implements IClientInvoker {
     httpExtension.setVerb(HttpVerbUtil.convertHttpVerbStringToNumber(method));
 
     const msgSerialized = new Any();
-    msgSerialized.setValue(Buffer.from(JSON.stringify(data), "utf-8"));
+    const {serializedData, contentType} = SerializerUtil.serializeGrpc(data);
+    msgSerialized.setValue(serializedData);
 
     const msgInvoke = new InvokeRequest();
     msgInvoke.setMethod(methodName);
     msgInvoke.setHttpExtension(httpExtension);
     msgInvoke.setData(msgSerialized);
-    msgInvoke.setContentType("application/json");
+    msgInvoke.setContentType(contentType);
 
     msgInvokeService.setMessage(msgInvoke);
 

src/implementation/Client/GRPCClient/pubsub.ts

@@ -15,6 +15,7 @@ import GRPCClient from './GRPCClient';
 import { PublishEventRequest } from "../../../proto/dapr/proto/runtime/v1/dapr_pb";
 import IClientPubSub from "../../../interfaces/Client/IClientPubSub";
 import { Logger } from '../../../logger/Logger';
+import * as SerializerUtil from "../../../utils/Serializer.util";
 
 // https://docs.dapr.io/reference/api/pubsub_api/
 export default class GRPCClientPubSub implements IClientPubSub {
@@ -32,7 +33,7 @@ export default class GRPCClientPubSub implements IClientPubSub {
     const msgService = new PublishEventRequest();
     msgService.setPubsubName(pubSubName);
     msgService.setTopic(topic);
-    msgService.setData(Buffer.from(JSON.stringify(data), "utf-8"));
+    msgService.setData(SerializerUtil.serializeGrpc(data).serializedData);
 
     return new Promise((resolve, reject) => {
       const client = this.client.getClient();

src/utils/Serializer.util.ts

@@ -0,0 +1,23 @@
+/*
+Copyright 2022 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+export function serializeGrpc(data: any): {serializedData: Buffer, contentType: string} {
+    let serializedData = data;
+    let contentType = "application/octet-stream";
+    if (!(data instanceof Buffer)) {
+        serializedData = Buffer.from(JSON.stringify(data), "utf-8");
+        contentType = "application/json";
+    }
+
+    return {serializedData, contentType};
+}

test/unit/utils/Serializer.util.test.ts

@@ -0,0 +1,27 @@
+/*
+Copyright 2022 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import * as SerializerUtil from "../../../src/utils/Serializer.util"
+
+describe('serializer', () => {
+    it('Object should be serialized to Buffer', () => {
+        const data = SerializerUtil.serializeGrpc({ Hello: 'World' });
+        expect(Buffer.compare(data.serializedData, Buffer.from(JSON.stringify({ Hello: 'World' })))).toEqual(0);
+        expect(data.contentType).toEqual("application/json");
+    });
+    it('Buffer object should not be serialized again', () => {
+        const data = SerializerUtil.serializeGrpc(Buffer.from('Hello World'));
+        expect(Buffer.compare(data.serializedData, Buffer.from('Hello World'))).toEqual(0);
+        expect(data.contentType).toEqual("application/octet-stream");
+    });
+});