[Service Bus] Remove Connection Context from ServiceBusMessage (Azure#12398)

In Azure#12216, we moved the code to settle a message and renew its lock from the ServiceBusMessageImpl to the ServiceBusReceiver. After this move, the only dependency left on the ConnectionContext in the ServiceBusMessageImpl is for the DataTransformer. This PR updates the ServiceBusMessage constructor to take the DataTransformer instead of the ConnectionContext and thus remove this dependency altogether. 

This closes Azure#9944 as we no longer need to track the context or the receiver on the message.
This closes Azure#10620 as well. Since the context is not being tracked on the message anymore, it does not matter that the data transformer is tracked on the context.

Author: ramya-rao-a

sdk/servicebus/service-bus/src/core/batchingReceiver.ts

@@ -253,7 +253,7 @@ export class BatchingReceiverLite {
 
     this._createServiceBusMessage = (context: MessageAndDelivery) => {
       return new ServiceBusMessageImpl(
-        _connectionContext,
+        _connectionContext.dataTransformer,
         context.message!,
         context.delivery!,
         true,

sdk/servicebus/service-bus/src/core/managementClient.ts

@@ -802,7 +802,7 @@ export class ManagementClient extends LinkEntity<RequestResponseLink> {
       for (const msg of messages) {
         const decodedMessage = RheaMessageUtil.decode(msg.message);
         const message = new ServiceBusMessageImpl(
-          this._context,
+          this._context.dataTransformer,
           decodedMessage as any,
           { tag: msg["lock-token"] } as any,
           false,

sdk/servicebus/service-bus/src/core/streamingReceiver.ts

@@ -222,7 +222,7 @@ export class StreamingReceiver extends MessageReceiver {
       }
 
       const bMessage: ServiceBusMessageImpl = new ServiceBusMessageImpl(
-        this._context,
+        this._context.dataTransformer,
         context.message!,
         context.delivery!,
         true,

sdk/servicebus/service-bus/src/serviceBusMessage.ts

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { AmqpAnnotatedMessage, Constants } from "@azure/core-amqp";
+import { AmqpAnnotatedMessage, Constants, DataTransformer } from "@azure/core-amqp";
 import { Buffer } from "buffer";
 import Long from "long";
 import {
@@ -11,7 +11,6 @@ import {
   uuid_to_string,
   Message as RheaMessage
 } from "rhea-promise";
-import { ConnectionContext } from "./connectionContext";
 import { messageLogger as logger } from "./log";
 import { ReceiveMode } from "./models";
 import { reorderLockToken } from "./util/utils";
@@ -770,7 +769,7 @@ export class ServiceBusMessageImpl implements ServiceBusReceivedMessage {
    * @internal
    */
   constructor(
-    private readonly _context: ConnectionContext,
+    private readonly _dataTransformer: DataTransformer,
     msg: RheaMessage,
     delivery: Delivery,
     shouldReorderLockToken: boolean,
@@ -783,7 +782,7 @@ export class ServiceBusMessageImpl implements ServiceBusReceivedMessage {
       this.lockToken = undefined;
     }
     if (msg.body) {
-      this.body = this._context.dataTransformer.decode(msg.body);
+      this.body = this._dataTransformer.decode(msg.body);
     }
     // TODO: _amqpAnnotatedMessage is already being populated in fromRheaMessage(), no need to do it twice
     this._amqpAnnotatedMessage = AmqpAnnotatedMessage.fromRheaMessage(msg);

sdk/servicebus/service-bus/src/session/messageSession.ts

@@ -621,7 +621,7 @@ export class MessageSession extends LinkEntity<Receiver> {
         }
 
         const bMessage = new ServiceBusMessageImpl(
-          this._context,
+          this._context.dataTransformer,
           context.message!,
           context.delivery!,
           true,

sdk/servicebus/service-bus/test/internal/serviceBusMessage.spec.ts

@@ -2,7 +2,6 @@
 // Licensed under the MIT license.
 
 import { ServiceBusMessageImpl } from "../../src/serviceBusMessage";
-import { ConnectionContext } from "../../src/connectionContext";
 import {
   Delivery,
   uuid_to_string,
@@ -11,15 +10,13 @@ import {
   Message as RheaMessage
 } from "rhea-promise";
 import chai from "chai";
-import { Constants } from "@azure/core-amqp";
+import { Constants, DataTransformer } from "@azure/core-amqp";
 const assert = chai.assert;
 
-const fakeContext = {
-  dataTransformer: {
-    encode: (data) => data,
-    decode: (data) => data
-  }
-} as ConnectionContext;
+const fakeDataTransformer = {
+  encode: (data) => data,
+  decode: (data) => data
+} as DataTransformer;
 const fakeDelivery = {} as Delivery;
 
 describe("ServiceBusMessageImpl LockToken unit tests", () => {
@@ -38,7 +35,7 @@ describe("ServiceBusMessageImpl LockToken unit tests", () => {
 
   it("Lock token in peekLock mode", () => {
     const sbMessage = new ServiceBusMessageImpl(
-      fakeContext,
+      fakeDataTransformer,
       amqpMessage,
       { tag: fakeDeliveryTag } as Delivery,
       false,
@@ -50,7 +47,7 @@ describe("ServiceBusMessageImpl LockToken unit tests", () => {
 
   it("Lock token in receiveAndDelete mode", () => {
     const sbMessage = new ServiceBusMessageImpl(
-      fakeContext,
+      fakeDataTransformer,
       amqpMessage,
       { tag: fakeDeliveryTag } as Delivery,
       false,
@@ -99,7 +96,7 @@ describe("ServiceBusMessageImpl AmqpAnnotations unit tests", () => {
   };
 
   const sbMessage = new ServiceBusMessageImpl(
-    fakeContext,
+    fakeDataTransformer,
     amqpMessage,
     fakeDelivery,
     false,