[core-amqp][event-hubs][service-bus] move DataTransformer from core-amqp to client packages (Azure#12415)

Part of the list of breaking changes to core-amqp v2 in Azure#12116
Replaces Azure#12320 (precipitated by Azure#12320 (comment))

This change moves the `DataTransformer` interface and `DefaultDataTransformer` class to service-bus and event-hub packages.

When we establish what our data serde strategy is, we can revisit using a shared common serde solution.

Author: chradek

sdk/core/core-amqp/CHANGELOG.md

@@ -2,8 +2,11 @@
 
 ## 2.0.0 (Unreleased)
 
-### Breaking Changes
+### Breaking changes
 
+- Continuing our work to clean the public API surface that we started in 2.0.0-beta.1, `DataTransformer` and `DefaultDataTransformer` are no longer exported. 
+  `dataTransformer` has been removed from `ConnectionContextBase` and `ConnectionContextBaseParameters`.
+  This allows us to consider other forms of implementing serializers in the future.
 - Previously, `ConnectionConfig.validate()` overridden entityPath if `undefined` with `String(undefined) = "undefined"`. This has been updated to retain `undefined` in the validation.
   [PR 12321](https://github.com/Azure/azure-sdk-for-js/pull/12321)
 

sdk/core/core-amqp/review/core-amqp.api.md

@@ -232,7 +232,6 @@ export interface ConnectionContextBase {
     connection: Connection;
     connectionId: string;
     connectionLock: string;
-    dataTransformer: DataTransformer;
     negotiateClaimLock: string;
     refreshConnection: () => void;
     wasConnectionCloseCalled: boolean;
@@ -373,23 +372,10 @@ export const Constants: {
 export interface CreateConnectionContextBaseParameters {
     config: ConnectionConfig;
     connectionProperties: ConnectionProperties;
-    dataTransformer?: DataTransformer;
     isEntityPathRequired?: boolean;
     operationTimeoutInMs?: number;
 }
 
-// @public
-export interface DataTransformer {
-    decode: (body: any) => any;
-    encode: (body: any) => any;
-}
-
-// @public
-export class DefaultDataTransformer implements DataTransformer {
-    decode(body: any): any;
-    encode(body: any): any;
-}
-
 // @public
 export const defaultLock: AsyncLock;
 

sdk/core/core-amqp/src/ConnectionContextBase.ts

@@ -3,7 +3,6 @@
 
 import { Connection, ConnectionOptions, generate_uuid } from "rhea-promise";
 import { CbsClient } from "./cbs";
-import { DataTransformer, DefaultDataTransformer } from "./dataTransformer";
 import { ConnectionConfig } from "./connectionConfig/connectionConfig";
 
 import { Constants } from "./util/constants";
@@ -44,12 +43,6 @@ export interface ConnectionContextBase {
    * called on the connection object.
    */
   wasConnectionCloseCalled: boolean;
-  /**
-   * @property {DataTransformer} dataTransformer A DataTransformer object that has methods named
-   * - encode Responsible for encoding the AMQP message before sending it on the wire.
-   * - decode Responsible for decoding the received AMQP message before passing it to the customer.
-   */
-  dataTransformer: DataTransformer;
   /**
    * @property {CbsClient} cbsSession A reference to the cbs session ($cbs endpoint) on the
    * underlying AMQP connection for the EventHub Client.
@@ -95,11 +88,6 @@ export interface CreateConnectionContextBaseParameters {
    * the AMQP connection.
    */
   connectionProperties: ConnectionProperties;
-  /**
-   * @property {DataTransformer} [dataTransformer] The datatransformer to be used for encoding and
-   * decoding messages. Default value: DefaultDataTransformer
-   */
-  dataTransformer?: DataTransformer;
   /**
    * @property {boolean} [isEntityPathRequired] Determines whether entity path should be a part of
    * the connection config. If `true` it must be present, `false` otherwise. Default value false.
@@ -178,7 +166,6 @@ export const ConnectionContextBase = {
       connectionId: connection.id,
       cbsSession: new CbsClient(connection, connectionLock),
       config: parameters.config,
-      dataTransformer: parameters.dataTransformer || new DefaultDataTransformer(),
       refreshConnection() {
         const connection = new Connection(connectionOptions);
         const connectionLock = `${Constants.establishConnection}-${generate_uuid()}`;

sdk/core/core-amqp/src/index.ts

@@ -5,7 +5,6 @@
 
 export { RequestResponseLink, SendRequestOptions } from "./requestResponseLink";
 export { retry, RetryOptions, RetryConfig, RetryOperationType, RetryMode } from "./retry";
-export { DataTransformer, DefaultDataTransformer } from "./dataTransformer";
 export { TokenType } from "./auth/token";
 
 export { ConnectionConfig, ConnectionConfigOptions } from "./connectionConfig/connectionConfig";

sdk/core/core-amqp/test/context.spec.ts

@@ -3,7 +3,7 @@
 
 import * as chai from "chai";
 const should = chai.should();
-import { CbsClient, ConnectionConfig, ConnectionContextBase, DefaultDataTransformer } from "../src";
+import { CbsClient, ConnectionConfig, ConnectionContextBase } from "../src";
 import { Connection } from "rhea-promise";
 import { isNode } from "../src/util/utils";
 
@@ -26,14 +26,12 @@ describe("ConnectionContextBase", function() {
     should.exist(context.connectionId);
     should.exist(context.connectionLock);
     should.exist(context.negotiateClaimLock);
-    should.exist(context.dataTransformer);
     context.wasConnectionCloseCalled.should.equal(false);
     context.connection.should.instanceOf(Connection);
     context.connection.options.properties!.product.should.equal("MSJSClient");
     context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
     context.connection.options.properties!.version.should.equal("1.0.0");
     context.cbsSession.should.instanceOf(CbsClient);
-    context.dataTransformer.should.instanceOf(DefaultDataTransformer);
     done();
   });
 
@@ -133,7 +131,6 @@ describe("ConnectionContextBase", function() {
       should.exist(context.connectionId);
       should.exist(context.connectionLock);
       should.exist(context.negotiateClaimLock);
-      should.exist(context.dataTransformer);
       context.wasConnectionCloseCalled.should.equal(false);
       context.cbsSession.should.instanceOf(CbsClient);
 

sdk/eventhub/event-hubs/rollup.base.config.js

@@ -122,7 +122,9 @@ export function browserConfig(test = false) {
 
       cjs({
         namedExports: {
-          "@opentelemetry/api": ["CanonicalCode", "SpanKind", "TraceFlags"]
+          "@opentelemetry/api": ["CanonicalCode", "SpanKind", "TraceFlags"],
+          chai: ["should", "assert"],
+          assert: ["equal", "deepEqual", "notEqual"]
         }
       }),
 

sdk/core/core-amqp/src/dataTransformer.ts …enthub/event-hubs/src/dataTransformer.tssdk/core/core-amqp/src/dataTransformer.ts renamed to sdk/eventhub/event-hubs/src/dataTransformer.ts sdk/core/core-amqp/src/dataTransformer.ts renamed to sdk/eventhub/event-hubs/src/dataTransformer.ts

@@ -2,32 +2,16 @@
 // Licensed under the MIT license.
 
 import { message } from "rhea-promise";
-import { logErrorStackTrace, logger } from "./log";
 import isBuffer from "is-buffer";
 import { Buffer } from "buffer";
-
-/**
- * Describes the transformations that can be performed to encode/decode the data before sending it
- * on (or receiving it from) the wire.
- */
-export interface DataTransformer {
-  /**
-   * @property {Function} encode A function that takes the body property from an EventData object
-   * and returns an encoded body (some form of AMQP type).
-   */
-  encode: (body: any) => any;
-  /**
-   * @property {Function} decode A function that takes the body property from an AMQP message
-   * and returns the decoded message body. If it cannot decode the body then it returns the body
-   * as-is.
-   */
-  decode: (body: any) => any;
-}
+import { logErrorStackTrace, logger } from "./log";
 
 /**
  * The default data transformer that will be used by the Azure SDK.
+ * @internal
+ * @ingore
  */
-export class DefaultDataTransformer implements DataTransformer {
+export const defaultDataTransformer = {
   /**
    * A function that takes the body property from an EventData object
    * and returns an encoded body (some form of AMQP type).
@@ -62,7 +46,7 @@ export class DefaultDataTransformer implements DataTransformer {
       }
     }
     return result;
-  }
+  },
 
   /**
    * @property {Function} [decode] A function that takes the body property from an AMQP message
@@ -99,4 +83,4 @@ export class DefaultDataTransformer implements DataTransformer {
     }
     return processedBody;
   }
-}
+};

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

@@ -8,6 +8,7 @@ import { throwTypeErrorIfParameterMissing } from "./util/error";
 import { Span, SpanContext } from "@opentelemetry/api";
 import { TRACEPARENT_PROPERTY, instrumentEventData } from "./diagnostics/instrumentEventData";
 import { createMessageSpan } from "./diagnostics/messageSpan";
+import { defaultDataTransformer } from "./dataTransformer";
 
 /**
  * The amount of bytes to reserve as overhead for a small message.
@@ -303,7 +304,7 @@ export class EventDataBatchImpl implements EventDataBatch {
 
     // Convert EventData to RheaMessage.
     const amqpMessage = toRheaMessage(eventData, this._partitionKey);
-    amqpMessage.body = this._context.dataTransformer.encode(eventData.body);
+    amqpMessage.body = defaultDataTransformer.encode(eventData.body);
     const encodedMessage = message.encode(amqpMessage);
 
     let currentSize = this._sizeInBytes;

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

@@ -25,6 +25,7 @@ import { ConnectionContext } from "./connectionContext";
 import { LinkEntity } from "./linkEntity";
 import { EventPosition, getEventPositionFilter } from "./eventPosition";
 import { AbortError, AbortSignalLike } from "@azure/abort-controller";
+import { defaultDataTransformer } from "./dataTransformer";
 
 /**
  * @ignore
@@ -230,7 +231,7 @@ export class EventHubReceiver extends LinkEntity {
 
     const data: EventDataInternal = fromRheaMessage(context.message);
     const receivedEventData: ReceivedEventData = {
-      body: this._context.dataTransformer.decode(context.message.body),
+      body: defaultDataTransformer.decode(context.message.body),
       properties: data.properties,
       offset: data.offset!,
       sequenceNumber: data.sequenceNumber!,

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

@@ -31,6 +31,7 @@ import { SendOptions } from "./models/public";
 import { getRetryAttemptTimeoutInMs } from "./util/retries";
 import { AbortError, AbortSignalLike } from "@azure/abort-controller";
 import { EventDataBatch, isEventDataBatch } from "./eventDataBatch";
+import { defaultDataTransformer } from "./dataTransformer";
 
 /**
  * Describes the EventHubSender that will send event data to EventHub.
@@ -322,7 +323,7 @@ export class EventHubSender extends LinkEntity {
         // Convert EventData to RheaMessage.
         for (let i = 0; i < events.length; i++) {
           const message = toRheaMessage(events[i], partitionKey);
-          message.body = this._context.dataTransformer.encode(events[i].body);
+          message.body = defaultDataTransformer.encode(events[i].body);
           messages[i] = message;
         }
         // Encode every amqp message and then convert every encoded message to amqp data section