[@azure/event-hubs] Adding disableDeserialization option when subscribing

sdk/eventhub/event-hubs/CHANGELOG.md

@@ -1,8 +1,9 @@
 # Release History
 
-## 5.6.1 (Unreleased)
+## 5.7.0 (Unreleased)
 
 ### Features Added
+- Added `skipParsingBodyAsJson` optional parameter to `EventHubConsumerClient.subscribe` method. When set to `true` it will disable the client from running `JSON.parse()` on the message body when receiving the message. Not applicable if the message was sent with AMQP body type `value` or `sequence`.
 
 ### Breaking Changes
 

sdk/eventhub/event-hubs/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/event-hubs",
   "sdk-type": "client",
-  "version": "5.6.1",
+  "version": "5.7.0",
   "description": "Azure Event Hubs SDK for JS.",
   "author": "Microsoft Corporation",
   "license": "MIT",

sdk/eventhub/event-hubs/review/event-hubs.api.md

@@ -277,6 +277,7 @@ export interface SubscribeOptions {
     maxBatchSize?: number;
     maxWaitTimeInSeconds?: number;
     ownerLevel?: number;
+    skipParsingBodyAsJson?: boolean;
     startPosition?: EventPosition | {
         [partitionId: string]: EventPosition;
     };

sdk/eventhub/event-hubs/src/dataTransformer.ts

@@ -75,22 +75,29 @@ export const defaultDataTransformer = {
    * indicating which part of the AMQP message the body was decoded from.
    *
    * @param body - The AMQP message body as received from rhea.
+   * @param skipParsingBodyAsJson - Boolean to skip running JSON.parse() on message body when body type is `content`.
    * @returns The decoded/raw body and the body type.
    */
-  decode(body: unknown | RheaAmqpSection): { body: unknown; bodyType: BodyTypes } {
+  decode(
+    body: unknown | RheaAmqpSection,
+    skipParsingBodyAsJson: boolean
+  ): { body: unknown; bodyType: BodyTypes } {
     try {
       if (isRheaAmqpSection(body)) {
         switch (body.typecode) {
           case dataSectionTypeCode:
-            return { body: tryToJsonDecode(body.content), bodyType: "data" };
+            return {
+              body: skipParsingBodyAsJson ? body.content : tryToJsonDecode(body.content),
+              bodyType: "data"
+            };
           case sequenceSectionTypeCode:
             return { body: body.content, bodyType: "sequence" };
           case valueSectionTypeCode:
             return { body: body.content, bodyType: "value" };
         }
       } else {
         if (isBuffer(body)) {
-          return { body: tryToJsonDecode(body), bodyType: "data" };
+          return { body: skipParsingBodyAsJson ? body : tryToJsonDecode(body), bodyType: "data" };
         }
 
         return { body, bodyType: "value" };

sdk/eventhub/event-hubs/src/eventData.ts

@@ -156,11 +156,15 @@ const messagePropertiesMap = {
 /**
  * Converts the AMQP message to an EventData.
  * @param msg - The AMQP message that needs to be converted to EventData.
+ * @param skipParsingBodyAsJson - Boolean to skip running JSON.parse() on message body when body type is `content`.
  * @hidden
  */
-export function fromRheaMessage(msg: RheaMessage): EventDataInternal {
+export function fromRheaMessage(
+  msg: RheaMessage,
+  skipParsingBodyAsJson: boolean
+): EventDataInternal {
   const rawMessage = AmqpAnnotatedMessage.fromRheaMessage(msg);
-  const { body, bodyType } = defaultDataTransformer.decode(msg.body);
+  const { body, bodyType } = defaultDataTransformer.decode(msg.body, skipParsingBodyAsJson);
   rawMessage.bodyType = bodyType;
 
   const data: EventDataInternal = {

sdk/eventhub/event-hubs/src/eventHubConsumerClientModels.ts

@@ -198,6 +198,12 @@ export interface SubscribeOptions {
    * Options for configuring tracing.
    */
   tracingOptions?: OperationTracingOptions;
+  /**
+   * Option to disable the client from running JSON.parse() on the message body when receiving the message.
+   * Not applicable if the message was sent with AMQP body type value or sequence. Use this option when you
+   * prefer to work directly with the bytes present in the message body than have the client attempt to parse it.
+   */
+  skipParsingBodyAsJson?: boolean;
 }
 
 /**

sdk/eventhub/event-hubs/src/eventHubReceiver.ts

@@ -221,8 +221,10 @@ export class EventHubReceiver extends LinkEntity {
     if (!context.message) {
       return;
     }
-
-    const data: EventDataInternal = fromRheaMessage(context.message);
+    const data: EventDataInternal = fromRheaMessage(
+      context.message,
+      !!this.options.skipParsingBodyAsJson
+    );
     const rawMessage = data.getRawAmqpMessage();
     const receivedEventData: ReceivedEventData = {
       body: data.body,

sdk/eventhub/event-hubs/src/models/private.ts

@@ -76,14 +76,18 @@ export interface CommonEventProcessorOptions
  * consumers to fail if their `ownerLevel` is lower or doesn't exist.
  * - `retryOptions`: The retry options used to govern retry attempts when an issue is encountered while receiving events.
  * A simple usage can be `{ "maxRetries": 4 }`.
+ * - `skipParsingBodyAsJson` : Option to disable the client from running JSON.parse() on the message body when receiving the message.
+ * Not applicable if the message was sent with AMQP body type value or sequence. Use this option when you prefer to work directly with
+ * the bytes present in the message body than have the client attempt to parse it.
  *
  * Example usage:
  * ```js
  * {
  *     retryOptions: {
  *         maxRetries: 4
  *     },
- *     trackLastEnqueuedEventProperties: false
+ *     trackLastEnqueuedEventProperties: false,
+ *     skipParsingBodyAsJson: true
  * }
  * ```
  * @internal
@@ -113,4 +117,10 @@ export interface EventHubConsumerOptions {
    * against periodically making requests for partition properties using the Event Hub client.
    */
   trackLastEnqueuedEventProperties?: boolean;
+  /**
+   * Option to disable the client from running JSON.parse() on the message body when receiving the message.
+   * Not applicable if the message was sent with AMQP body type value or sequence. Use this option when you
+   * prefer to work directly with the bytes present in the message body than have the client attempt to parse it.
+   */
+  skipParsingBodyAsJson?: boolean;
 }

sdk/eventhub/event-hubs/src/partitionPump.ts

@@ -90,7 +90,8 @@ export class PartitionPump {
       {
         ownerLevel: this._processorOptions.ownerLevel,
         trackLastEnqueuedEventProperties: this._processorOptions.trackLastEnqueuedEventProperties,
-        retryOptions: this._processorOptions.retryOptions
+        retryOptions: this._processorOptions.retryOptions,
+        skipParsingBodyAsJson: this._processorOptions.skipParsingBodyAsJson
       }
     );
 

sdk/eventhub/event-hubs/src/util/constants.ts

@@ -6,5 +6,5 @@
  */
 export const packageJsonInfo = {
   name: "@azure/event-hubs",
-  version: "5.6.1"
+  version: "5.7.0"
 };

sdk/eventhub/event-hubs/test/internal/amqp.spec.ts

@@ -14,23 +14,29 @@ testWithServiceTypes(() => {
       assert.isFalse(isAmqpAnnotatedMessage({ body: "hello world" }));
       assert.isFalse(
         isAmqpAnnotatedMessage(
-          fromRheaMessage({
-            message_annotations: {
-              [Constants.enqueuedTime]: Date.now()
+          fromRheaMessage(
+            {
+              message_annotations: {
+                [Constants.enqueuedTime]: Date.now()
+              },
+              body: undefined
             },
-            body: undefined
-          })
+            false
+          )
         )
       );
 
       assert.isTrue(
         isAmqpAnnotatedMessage(
-          fromRheaMessage({
-            message_annotations: {
-              [Constants.enqueuedTime]: Date.now()
+          fromRheaMessage(
+            {
+              message_annotations: {
+                [Constants.enqueuedTime]: Date.now()
+              },
+              body: undefined
             },
-            body: undefined
-          }).getRawAmqpMessage()
+            false
+          ).getRawAmqpMessage()
         )
       );