[@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 `skipJsonParsingContent` optional parameter to `EventHubConsumerClient.subscribe` method. When se to `true` it will disable all automatic `JSON.parse` when the body of an event (`event body` or `event.body.content`) is a stringified JSON.
 
 ### 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;
+    skipJsonParsingContent?: 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 skipJsonParsingContent - Boolean to disable automatic JSON parsing when receiving a string in the event body.
    * @returns The decoded/raw body and the body type.
    */
-  decode(body: unknown | RheaAmqpSection): { body: unknown; bodyType: BodyTypes } {
+  decode(
+    body: unknown | RheaAmqpSection,
+    skipJsonParsingContent: boolean
+  ): { body: unknown; bodyType: BodyTypes } {
     try {
       if (isRheaAmqpSection(body)) {
         switch (body.typecode) {
           case dataSectionTypeCode:
-            return { body: tryToJsonDecode(body.content), bodyType: "data" };
+            return {
+              body: skipJsonParsingContent ? 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: skipJsonParsingContent ? 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 skipJsonParsingContent - Boolean to disable automatic JSON parsing when receiving a string in the event body.
  * @hidden
  */
-export function fromRheaMessage(msg: RheaMessage): EventDataInternal {
+export function fromRheaMessage(
+  msg: RheaMessage,
+  skipJsonParsingContent: boolean
+): EventDataInternal {
   const rawMessage = AmqpAnnotatedMessage.fromRheaMessage(msg);
-  const { body, bodyType } = defaultDataTransformer.decode(msg.body);
+  const { body, bodyType } = defaultDataTransformer.decode(msg.body, skipJsonParsingContent);
   rawMessage.bodyType = bodyType;
 
   const data: EventDataInternal = {

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

@@ -198,6 +198,10 @@ export interface SubscribeOptions {
    * Options for configuring tracing.
    */
   tracingOptions?: OperationTracingOptions;
+  /**
+   * Optional boolean to disable automatic JSON parsing when receiving a string in the event body.
+   */
+  skipJsonParsingContent?: 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.skipJsonParsingContent
+    );
     const rawMessage = data.getRawAmqpMessage();
     const receivedEventData: ReceivedEventData = {
       body: data.body,

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

@@ -76,6 +76,7 @@ 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 }`.
+ * - `skipJsonParsingContent` : Optional boolean to disable automatic JSON parsing when receiving a string in the event body.
  *
  * Example usage:
  * ```js
@@ -113,4 +114,8 @@ export interface EventHubConsumerOptions {
    * against periodically making requests for partition properties using the Event Hub client.
    */
   trackLastEnqueuedEventProperties?: boolean;
+  /**
+   * Optional boolean to disable automatic JSON parsing when receiving a string in the event body.
+   */
+  skipJsonParsingContent?: 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,
+        skipJsonParsingContent: this._processorOptions.skipJsonParsingContent
       }
     );
 

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()
         )
       );