[Service Bus] Enable sending large messages (Azure#23014)

* new test

* fix sending large messages

* formatting and docs

* format

* changelog

* more docs

* Update sdk/servicebus/service-bus/CHANGELOG.md

Co-authored-by: Jeremy Meng <yumeng@microsoft.com>

* Update sdk/servicebus/service-bus/test/internal/sendBatch.spec.ts

* Apply suggestions from code review

* fix error message

Co-authored-by: Jeremy Meng <yumeng@microsoft.com>

Author: HarshaNalluru

sdk/servicebus/service-bus/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Updating the `sendMessages` API to not batch the messages when it is just a single message. This allows users to send individual large message (>1MB) using the `sendMessages` API. [#23014](https://github.com/Azure/azure-sdk-for-js/pull/23014)
+
 ### Other Changes
 
 ## 7.7.0 (2022-08-09)

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

@@ -25,9 +25,9 @@ import {
 import { OperationOptionsBase } from "./modelsToBeSharedWithEventHubs";
 import { TracingSpanLink } from "@azure/core-tracing";
 import { senderLogger as logger } from "./log";
-import { ServiceBusError } from "./serviceBusError";
 import { toSpanOptions, tracingClient } from "./diagnostics/tracing";
 import { ensureValidIdentifier } from "./util/utils";
+import { ServiceBusError } from "./serviceBusError";
 
 /**
  * A Sender can be used to send messages, schedule messages to be sent at a later time
@@ -49,6 +49,14 @@ export interface ServiceBusSender {
    * - All messages passed to the same sendMessages() call should have the same `sessionId` (if using
    *  sessions) and the same `partitionKey` (if using partitions).
    *
+   * **Note:**
+   *
+   *    __If you want to send messages of size greater than 1MB, please send individual messages instead of sending a batched message or an array of messages like below.__
+   *
+   *  `await sender.sendMessages(message);`
+   *
+   * __This is because the batched messages are not capable of sending the larger messages yet. You'll hit the `force detached` error in this case otherwise. Read [service-bus-premium-messaging#large-messages-support](https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-premium-messaging#large-messages-support). More info at [#23014](https://github.com/Azure/azure-sdk-for-js/pull/23014).__
+   *
    * @param messages - A single message or an array of messages or a batch of messages created via the createBatch()
    * method to send.
    * @param options - Options bag to pass an abort signal or tracing options.
@@ -202,13 +210,28 @@ export class ServiceBusSenderImpl implements ServiceBusSender {
     this._throwIfSenderOrConnectionClosed();
     throwTypeErrorIfParameterMissing(this._context.connectionId, "messages", messages);
 
+    if (!isServiceBusMessageBatch(messages) && !Array.isArray(messages)) {
+      // Case 1: Single message
+      throwIfNotValidServiceBusMessage(messages, errorInvalidMessageTypeSingleOrArray);
+      return tracingClient.withSpan(
+        "ServiceBusSender.send",
+        options ?? {},
+        (updatedOptions) => this._sender.send(messages, updatedOptions),
+        {
+          ...toSpanOptions(
+            { entityPath: this.entityPath, host: this._context.config.host },
+            "client"
+          ),
+        }
+      );
+    }
+
     let batch: ServiceBusMessageBatch;
     if (isServiceBusMessageBatch(messages)) {
+      // Case 2: Batch message
       batch = messages;
     } else {
-      if (!Array.isArray(messages)) {
-        messages = [messages];
-      }
+      // Case 3: Array of messages
       batch = await this.createMessageBatch(options);
       for (const message of messages) {
         throwIfNotValidServiceBusMessage(message, errorInvalidMessageTypeSingleOrArray);

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

@@ -34,7 +34,9 @@ describe("Send Batch", () => {
   });
 
   async function beforeEachTest(entityType: TestClientType): Promise<void> {
-    entityNames = await serviceBusClient.test.createTestEntities(entityType);
+    entityNames = await serviceBusClient.test.createTestEntities(entityType, {
+      maxMessageSizeInKilobytes: 102400,
+    });
 
     sender = serviceBusClient.test.addToCleanup(
       serviceBusClient.createSender(entityNames.queue ?? entityNames.topic!)
@@ -258,6 +260,37 @@ describe("Send Batch", () => {
     });
   });
 
+  describe("Send single message - size > 1 MB", function (): void {
+    afterEach(async () => {
+      await afterEachTest();
+    });
+
+    function prepareMessage(useSessions: boolean): ServiceBusMessage {
+      return {
+        body: Buffer.alloc(1024 * 1024),
+        sessionId: useSessions ? `s` : undefined,
+      };
+    }
+
+    async function testSend(): Promise<void> {
+      // Prepare messages to send
+      const messageToSend = prepareMessage(entityNames.usesSessions);
+      await sender.sendMessages(messageToSend);
+      // receive all the messages in receive and delete mode
+      await serviceBusClient.test.verifyAndDeleteAllSentMessages(entityNames, [messageToSend]);
+    }
+
+    it(`${noSessionTestClientType}: SendBatch`, async function (): Promise<void> {
+      await beforeEachTest(noSessionTestClientType);
+      await testSend();
+    });
+
+    it(`${withSessionTestClientType}: SendBatch`, async function (): Promise<void> {
+      await beforeEachTest(withSessionTestClientType);
+      await testSend();
+    });
+  });
+
   describe("Send multiple heterogenous messages - size > max_batch_size_allowed", function (): void {
     afterEach(async () => {
       await afterEachTest();

sdk/servicebus/service-bus/test/public/utils/testutils2.ts

@@ -93,7 +93,8 @@ function getEntityNames(testClientType: TestClientType): AutoGeneratedEntity {
 }
 
 async function createTestEntities(
-  testClientType: TestClientType
+  testClientType: TestClientType,
+  options?: { maxMessageSizeInKilobytes: number }
 ): Promise<ReturnType<typeof getEntityNames>> {
   const relatedEntities = getEntityNames(testClientType);
 
@@ -103,13 +104,15 @@ async function createTestEntities(
       enableBatchedOperations: true,
       enablePartitioning: relatedEntities.isPartitioned,
       requiresSession: relatedEntities.usesSessions,
+      maxMessageSizeInKilobytes: options?.maxMessageSizeInKilobytes,
     });
   }
 
   if (relatedEntities.topic) {
     await recreateTopic(relatedEntities.topic, {
       enablePartitioning: relatedEntities.isPartitioned,
       enableBatchedOperations: true,
+      maxMessageSizeInKilobytes: options?.maxMessageSizeInKilobytes,
     });
   }
 
@@ -315,14 +318,17 @@ export class ServiceBusTestHelpers {
    * entity from _this_ instance of the ServiceBusTestHelpers.
    */
   async createTestEntities(
-    testClientType: TestClientType
+    testClientType: TestClientType,
+    options?: {
+      maxMessageSizeInKilobytes: number;
+    }
   ): Promise<ReturnType<typeof getEntityNames>> {
     // TODO: for now these aren't randomly named. This is prep so we can
     // do that soon.
     let entityValues = this._testClientEntities.get(testClientType);
 
     if (entityValues == null) {
-      entityValues = await createTestEntities(testClientType);
+      entityValues = await createTestEntities(testClientType, options);
       this._testClientEntities.set(testClientType, entityValues);
     }