[service-bus] Temporarily disable AMQP body type encoding support for 7.0.6 (Azure#15184)

Disable amqp body type support until we release it in an official 'beta'.

Runtime prevention:
- Prevent isAmqpAnnotatedMessage() from ever returning true (disables all AMQP body type related code paths)
- Fix error message constants so they feature-switch and don't mention AmqpAnnotatedMessage

Compile time prevention:
- Remove AmqpAnnotatedMessage from the public facing types (ServiceBusMessageBatch and ServiceBusSender)

All tests and internals are intact and tests are enabling the code paths internally so tests are still working properly.

Author: richardpark-msft

sdk/servicebus/service-bus/review/service-bus.api.md

@@ -400,7 +400,7 @@ export interface ServiceBusMessageBatch {
     // @internal
     readonly _messageSpanContexts: SpanContext[];
     readonly sizeInBytes: number;
-    tryAddMessage(message: ServiceBusMessage | AmqpAnnotatedMessage, options?: TryAddOptions): boolean;
+    tryAddMessage(message: ServiceBusMessage, options?: TryAddOptions): boolean;
 }
 
 // @public
@@ -459,7 +459,7 @@ export interface ServiceBusSender {
     entityPath: string;
     isClosed: boolean;
     scheduleMessages(messages: ServiceBusMessage | ServiceBusMessage[], scheduledEnqueueTimeUtc: Date, options?: OperationOptionsBase): Promise<Long[]>;
-    sendMessages(messages: ServiceBusMessage | ServiceBusMessage[] | ServiceBusMessageBatch | AmqpAnnotatedMessage | AmqpAnnotatedMessage[], options?: OperationOptionsBase): Promise<void>;
+    sendMessages(messages: ServiceBusMessage | ServiceBusMessage[] | ServiceBusMessageBatch, options?: OperationOptionsBase): Promise<void>;
 }
 
 // @public

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

@@ -51,12 +51,7 @@ export interface ServiceBusSender {
    * @throws `ServiceBusError` if the service returns an error while sending messages to the service.
    */
   sendMessages(
-    messages:
-      | ServiceBusMessage
-      | ServiceBusMessage[]
-      | ServiceBusMessageBatch
-      | AmqpAnnotatedMessage
-      | AmqpAnnotatedMessage[],
+    messages: ServiceBusMessage | ServiceBusMessage[] | ServiceBusMessageBatch,
     options?: OperationOptionsBase
   ): Promise<void>;
 

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

@@ -620,11 +620,33 @@ export function isServiceBusMessage(possible: unknown): possible is ServiceBusMe
   return isObjectWithProperties(possible, ["body"]);
 }
 
+let _featureAmqpBodyTypeEnabled = false;
+
+/**
+ * Disables preview AMQP body type support.
+ * @internal
+ */
+export function getFeatureAmqpBodyTypeEnabled(): boolean {
+  return _featureAmqpBodyTypeEnabled;
+}
+
+/**
+ * Enables preview AMQP body type support.
+ * @internal
+ */
+export function setFeatureAmqpBodyTypeEnabledForTesting(enable: boolean): void {
+  _featureAmqpBodyTypeEnabled = enable;
+}
+
 /**
  * @internal
  * @ignore
  */
 export function isAmqpAnnotatedMessage(possible: unknown): possible is AmqpAnnotatedMessage {
+  if (!_featureAmqpBodyTypeEnabled) {
+    return false;
+  }
+
   return (
     isObjectWithProperties(possible, ["body", "bodyType"]) &&
     possible.constructor.name !== ServiceBusMessageImpl.name

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

@@ -70,10 +70,7 @@ export interface ServiceBusMessageBatch {
    * @param message - The message to add to the batch.
    * @returns A boolean value indicating if the message has been added to the batch or not.
    */
-  tryAddMessage(
-    message: ServiceBusMessage | AmqpAnnotatedMessage,
-    options?: TryAddOptions
-  ): boolean;
+  tryAddMessage(message: ServiceBusMessage, options?: TryAddOptions): boolean;
 
   /**
    * The AMQP message containing encoded events that were added to the batch.

sdk/servicebus/service-bus/src/util/errors.ts

@@ -5,6 +5,7 @@ import { logger, receiverLogger } from "../log";
 import Long from "long";
 import { ConnectionContext } from "../connectionContext";
 import {
+  getFeatureAmqpBodyTypeEnabled,
   isAmqpAnnotatedMessage,
   isServiceBusMessage,
   ServiceBusReceivedMessage
@@ -269,9 +270,11 @@ export function throwIfNotValidServiceBusMessage(
 }
 
 /** @internal */
-export const errorInvalidMessageTypeSingleOrArray =
-  "Provided value for 'messages' must be of type: ServiceBusMessage, AmqpAnnotatedMessage, ServiceBusMessageBatch or an array of type ServiceBusMessage or AmqpAnnotatedMessage.";
+export const errorInvalidMessageTypeSingleOrArray = getFeatureAmqpBodyTypeEnabled()
+  ? "Provided value for 'messages' must be of type: ServiceBusMessage, AmqpAnnotatedMessage, ServiceBusMessageBatch or an array of type ServiceBusMessage or AmqpAnnotatedMessage."
+  : "Provided value for 'messages' must be of type: ServiceBusMessage, ServiceBusMessageBatch or an array of type ServiceBusMessage";
 
 /** @internal */
-export const errorInvalidMessageTypeSingle =
-  "Provided value for 'message' must be of type: ServiceBusMessage or AmqpAnnotatedMessage.";
+export const errorInvalidMessageTypeSingle = getFeatureAmqpBodyTypeEnabled()
+  ? "Provided value for 'message' must be of type: ServiceBusMessage or AmqpAnnotatedMessage."
+  : "Provided value for 'message' must be of type: ServiceBusMessage";

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

@@ -2,12 +2,14 @@
 // Licensed under the MIT license.
 
 import * as chai from "chai";
-import { ServiceBusReceiver, ServiceBusSender } from "../../src";
-import {
-  createServiceBusClientForTests,
-  ServiceBusClientForTests
-} from "../public/utils/testutils2";
+import { ServiceBusSender } from "../../src";
 import { TestClientType } from "../public/utils/testUtils";
+import { ServiceBusSenderImpl } from "../../src/sender";
+import {
+  getFeatureAmqpBodyTypeEnabled,
+  setFeatureAmqpBodyTypeEnabledForTesting
+} from "../../src/serviceBusMessage";
+import { addServiceBusClientForLiveTesting } from "../public/utils/testutils2";
 const assert = chai.assert;
 
 describe("AMQP message encoding", () => {
@@ -16,35 +18,89 @@ describe("AMQP message encoding", () => {
 
   // Primitive types
   // http://docs.oasis-open.org/amqp/core/v1.0/csprd01/amqp-core-types-v1.0-csprd01.html#toc
+  const { sender: _senderImpl, receiver } = addServiceBusClientForLiveTesting(
+    TestClientType.UnpartitionedQueue
+  );
+  const sender: () => Omit<ServiceBusSender, "sendMessages"> &
+    Pick<ServiceBusSenderImpl, "sendMessages"> = () => {
+    return _senderImpl();
+  };
+
+  // NOTE: Will remove this entire test suiteonce we re-enable AMQP body type encoding.
+  describe("AMQP message encoding is disabled", () => {
+    ([
+      ["sequence", [1, 2, 3]],
+      ["value", "hello"],
+      ["data", "hello"]
+    ] as ["sequence" | "data" | "value", any][]).forEach(([originalBodyType, expectedBody]) => {
+      it("receive ServiceBusMessage and resend", async () => {
+        await (sender() as ServiceBusSender).sendMessages({
+          body: expectedBody,
+          // @ts-expect-error "`bodyType` is not a field in ServiceBusMessage. When AMQP body type encoding is disabled it won't compile"
+          bodyType: originalBodyType
+        });
 
-  describe("amqp encoding/decoding", () => {
-    let client: ServiceBusClientForTests;
-    let sender: ServiceBusSender;
-    let receiver: ServiceBusReceiver;
+        const messages = await receiver().receiveMessages(1);
+        const message = messages[0];
 
-    before(() => {
-      client = createServiceBusClientForTests();
+        assert.equal(
+          message._rawAmqpMessage.bodyType,
+          "data",
+          "Body type (when AMQP body type encoding is disabled) is always 'data'"
+        );
+
+        // now let's just resend it, unaltered
+        await sender().sendMessages(message);
+
+        const reencodedMessages = await receiver().receiveMessages(1);
+        const reencodedMessage = reencodedMessages[0];
+
+        assert.equal(reencodedMessage._rawAmqpMessage.bodyType, "data");
+        assert.deepEqual(reencodedMessage.body, expectedBody);
+      });
     });
 
-    beforeEach(async () => {
-      const testEntities = await client.test.createTestEntities(TestClientType.UnpartitionedQueue);
-      sender = await client.test.createSender(testEntities);
-      receiver = await client.test.createReceiveAndDeleteReceiver(testEntities);
+    it("ServiceBusMessageBatch.tryAdd()", async () => {
+      const batch = await sender().createMessageBatch();
+
+      assert.isTrue(
+        batch.tryAddMessage({
+          body: "hello",
+          // @ts-expect-error "`bodyType` is not a field in ServiceBusMessage. When AMQP body type encoding is disabled it won't compile"
+          bodyType: "value"
+        })
+      );
+
+      await sender().sendMessages(batch);
+
+      const reencodedMessages = await receiver().receiveMessages(1);
+      const reencodedMessage = reencodedMessages[0];
+
+      assert.equal(reencodedMessage._rawAmqpMessage.bodyType, "data");
+      assert.equal(reencodedMessage.body, "hello");
     });
+  });
 
-    afterEach(() => client.test.afterEach());
-    after(() => client.test.after());
+  describe("amqp encoding/decoding", () => {
+    beforeEach(() => {
+      const previousState = getFeatureAmqpBodyTypeEnabled();
+      assert.isFalse(previousState);
+      setFeatureAmqpBodyTypeEnabledForTesting(true);
+    });
+    afterEach(() => {
+      setFeatureAmqpBodyTypeEnabledForTesting(false);
+    });
 
     it("values", async () => {
       const valueTypes = [[1, 2, 3], 1, 1.5, "hello", { hello: "world" }];
 
       for (const valueType of valueTypes) {
-        await sender.sendMessages({
+        await sender().sendMessages({
           body: valueType,
           bodyType: "value"
         });
 
-        const messages = await receiver.receiveMessages(1);
+        const messages = await receiver().receiveMessages(1);
         const message = messages[0];
 
         assert.deepEqual(
@@ -67,12 +123,12 @@ describe("AMQP message encoding", () => {
       ];
 
       for (const sequenceType of sequenceTypes) {
-        await sender.sendMessages({
+        await sender().sendMessages({
           body: sequenceType,
           bodyType: "sequence"
         });
 
-        const messages = await receiver.receiveMessages(1);
+        const messages = await receiver().receiveMessages(1);
         const message = messages[0];
 
         assert.deepEqual(
@@ -94,12 +150,12 @@ describe("AMQP message encoding", () => {
       const dataTypes = [1, 1.5, "hello", { hello: "world" }, buff, [1, 2, 3]];
 
       for (const dataType of dataTypes) {
-        await sender.sendMessages({
+        await sender().sendMessages({
           body: dataType,
           bodyType: "data"
         });
 
-        const messages = await receiver.receiveMessages(1);
+        const messages = await receiver().receiveMessages(1);
         const message = messages[0];
 
         assert.deepEqual(
@@ -114,5 +170,35 @@ describe("AMQP message encoding", () => {
         );
       }
     });
+
+    ([
+      ["sequence", [1, 2, 3]],
+      ["value", "hello"],
+      ["data", "hello"]
+    ] as ["sequence" | "data" | "value", any][]).forEach(([expectedBodyType, expectedBody]) => {
+      it("receive ServiceBusMessage and resend", async () => {
+        // if we receive a message that was encoded to a non-data section
+        // and then re-send it (again, as a ServiceBusMessage) we should
+        // respect it.
+        await sender().sendMessages({
+          body: expectedBody,
+          bodyType: expectedBodyType
+        });
+
+        const messages = await receiver().receiveMessages(1);
+        const message = messages[0];
+
+        assert.equal(message._rawAmqpMessage.bodyType, expectedBodyType);
+
+        // now let's just resend it, unaltered
+        await sender().sendMessages(message);
+
+        const reencodedMessages = await receiver().receiveMessages(1);
+        const reencodedMessage = reencodedMessages[0];
+
+        assert.equal(reencodedMessage._rawAmqpMessage.bodyType, expectedBodyType);
+        assert.deepEqual(reencodedMessage.body, expectedBody);
+      });
+    });
   });
 });

sdk/servicebus/service-bus/test/internal/unit/amqpUnitTests.spec.ts

@@ -2,11 +2,13 @@
 // Licensed under the MIT license.
 
 import {
+  getFeatureAmqpBodyTypeEnabled,
   isAmqpAnnotatedMessage,
   isServiceBusMessage,
   ServiceBusMessage,
   ServiceBusMessageImpl,
   ServiceBusReceivedMessage,
+  setFeatureAmqpBodyTypeEnabledForTesting,
   toRheaMessage
 } from "../../../src/serviceBusMessage";
 import * as chai from "chai";
@@ -18,9 +20,33 @@ import {
   isRheaAmqpSection,
   valueSectionTypeCode
 } from "../../../src/dataTransformer";
+import {
+  errorInvalidMessageTypeSingle,
+  errorInvalidMessageTypeSingleOrArray
+} from "../../../src/util/errors";
 const assert = chai.assert;
 
 describe("AMQP message encoding", () => {
+  beforeEach(() => {
+    assert.equal(
+      "Provided value for 'messages' must be of type: ServiceBusMessage, ServiceBusMessageBatch or an array of type ServiceBusMessage",
+      errorInvalidMessageTypeSingleOrArray
+    );
+
+    assert.equal(
+      "Provided value for 'message' must be of type: ServiceBusMessage",
+      errorInvalidMessageTypeSingle
+    );
+
+    const previousState = getFeatureAmqpBodyTypeEnabled();
+    assert.isFalse(previousState);
+
+    setFeatureAmqpBodyTypeEnabledForTesting(true);
+  });
+  afterEach(() => {
+    setFeatureAmqpBodyTypeEnabledForTesting(false);
+  });
+
   const exampleReceivedMessage: () => ServiceBusReceivedMessage = () =>
     new ServiceBusMessageImpl(
       ({

sdk/servicebus/service-bus/test/public/amqpAnnotatedMessage.spec.ts

@@ -14,6 +14,10 @@ import {
 } from "../public/utils/testutils2";
 import { AmqpAnnotatedMessage } from "@azure/core-amqp";
 import { generateUuid } from "@azure/core-http";
+import {
+  getFeatureAmqpBodyTypeEnabled,
+  setFeatureAmqpBodyTypeEnabledForTesting
+} from "../../src/serviceBusMessage";
 
 const should = chai.should();
 chai.use(chaiAsPromised);
@@ -67,10 +71,15 @@ function getSampleAmqpAnnotatedMessage(randomTag?: string): AmqpAnnotatedMessage
 
 describe("AmqpAnnotatedMessage", function(): void {
   before(() => {
+    const previousState = getFeatureAmqpBodyTypeEnabled();
+    assert.isFalse(previousState);
+
+    setFeatureAmqpBodyTypeEnabledForTesting(true);
     serviceBusClient = createServiceBusClientForTests();
   });
 
   after(() => {
+    setFeatureAmqpBodyTypeEnabledForTesting(false);
     return serviceBusClient.test.after();
   });