[SB/EH] comment custom endpoint doc (Azure#38516)

* start

* custom endpoint over amqp

* fix

* config

* doc

* eventhub fix

* mypy

* custom port 2

* why blakc rynning here ok?

Author: l0lawrence

sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/pyproject.toml

@@ -7,3 +7,4 @@ pylint = true
 latestdependency = false
 mindependency = false
 whl_no_aio = false
+black = false

sdk/eventhub/azure-eventhub-checkpointstoreblob/pyproject.toml

@@ -4,3 +4,4 @@ pyright = false
 type_check_samples = false
 verifytypes = false
 pylint = true
+black = false

sdk/eventhub/azure-eventhub-checkpointstoretable/pyproject.toml

@@ -4,3 +4,4 @@ pyright = false
 type_check_samples = false
 verifytypes = false
 pylint = false
+black = false

sdk/eventhub/azure-eventhub/azure/eventhub/_configuration.py

@@ -29,7 +29,7 @@ def __init__(
         retry_backoff_max: int = 120,
         network_tracing: bool = False,
         http_proxy: Optional[Dict[str, Any]] = None,
-        transport_type: TransportType = TransportType.Amqp,
+        transport_type: Optional[TransportType] = None,
         auth_timeout: int = 60,
         prefetch: int = 300,
         max_batch_size: int = 300,
@@ -50,6 +50,8 @@ def __init__(
         self.network_tracing = network_tracing
         self.http_proxy = http_proxy
         self.transport_type = TransportType.AmqpOverWebsocket if self.http_proxy else transport_type
+        # if transport_type is not provided, it is None, we will default to Amqp
+        self.transport_type = self.transport_type or TransportType.Amqp
         self.auth_timeout = auth_timeout
         self.prefetch = prefetch
         self.max_batch_size = max_batch_size
@@ -81,9 +83,12 @@ def __init__(
             if self.custom_endpoint_address.find("//") == -1:
                 self.custom_endpoint_address = "sb://" + self.custom_endpoint_address
             endpoint = urlparse(self.custom_endpoint_address)
-            self.transport_type = TransportType.AmqpOverWebsocket
+            # this line is to maintain backward compatibility for custom endpoint
+            # if the transport_type is not provided, we will default to AmqpOverWebsocket
+            self.transport_type = transport_type or TransportType.AmqpOverWebsocket
             self.custom_endpoint_hostname = endpoint.hostname
             if amqp_transport.KIND == "pyamqp":
+                # if transport is amqp, we will not append this path to the final endpoint
                 self.custom_endpoint_address += "/$servicebus/websocket"
             # in case proxy and custom endpoint are both provided, we default port to 443 if it's not provided
             self.connection_port = endpoint.port or DEFAULT_AMQP_WSS_PORT

sdk/eventhub/azure-eventhub/azure/eventhub/_consumer_client.py

@@ -116,6 +116,7 @@ class EventHubConsumerClient(ClientBase):  # pylint: disable=client-accepts-api-
     :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
      the Event Hubs service, allowing network requests to be routed through any application gateways or
      other paths needed for the host environment. Default is None.
+     Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
      The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>".
      If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
     :paramtype custom_endpoint_address: str or None
@@ -307,6 +308,7 @@ def from_connection_string(
         :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
          the Event Hubs service, allowing network requests to be routed through any application gateways or
          other paths needed for the host environment. Default is None.
+         Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
          The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>".
          If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
         :paramtype custom_endpoint_address: str or None

sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py

@@ -122,6 +122,7 @@ class EventHubProducerClient(ClientBase):  # pylint: disable=client-accepts-api-
     :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
      the Event Hubs service, allowing network requests to be routed through any application gateways or
      other paths needed for the host environment. Default is None.
+     Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
      The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>".
      If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
     :paramtype custom_endpoint_address: Optional[str]
@@ -504,6 +505,7 @@ def from_connection_string(
         :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
          the Event Hubs service, allowing network requests to be routed through any application gateways or
          other paths needed for the host environment. Default is None.
+         Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
          The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>".
          If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
         :paramtype custom_endpoint_address: Optional[str]

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_connection.py

@@ -134,10 +134,15 @@ def __init__(  # pylint:disable=too-many-locals,too-many-statements
         # Custom Endpoint
         custom_endpoint_address = kwargs.get("custom_endpoint_address")
         custom_endpoint = None
+        custom_port = None
         if custom_endpoint_address:
             custom_parsed_url = urlparse(custom_endpoint_address)
-            custom_port = custom_parsed_url.port or WEBSOCKET_PORT
-            custom_endpoint = f"{custom_parsed_url.hostname}:{custom_port}{custom_parsed_url.path}"
+            if transport_type.value == TransportType.Amqp.value:
+                custom_port = custom_parsed_url.port or SECURE_PORT
+                custom_endpoint = f"{custom_parsed_url.hostname}"
+            else:
+                custom_port = custom_parsed_url.port or WEBSOCKET_PORT
+                custom_endpoint = f"{custom_parsed_url.hostname}:{custom_port}{custom_parsed_url.path}"
         self._container_id = container_id or str(uuid.uuid4())
         self._network_trace = network_trace
         self._network_trace_params = {"amqpConnection": self._container_id, "amqpSession": "", "amqpLink": ""}
@@ -163,6 +168,7 @@ def __init__(  # pylint:disable=too-many-locals,too-many-statements
                 credential=kwargs["sasl_credential"],
                 port=self._port,
                 custom_endpoint=custom_endpoint,
+                custom_port=custom_port,
                 socket_timeout=self._socket_timeout,
                 network_trace_params=self._network_trace_params,
                 **kwargs,

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_transport.py

@@ -491,9 +491,13 @@ class SSLTransport(_AbstractTransport):
 
     def __init__(self, host, *, port=AMQPS_PORT, socket_timeout=None, ssl_opts=None, **kwargs):
         self.sslopts = ssl_opts if isinstance(ssl_opts, dict) else {}
-        self.sslopts["server_hostname"] = host
+        self._custom_endpoint = kwargs.get("custom_endpoint")
+        self._custom_port = kwargs.get("custom_port")
+        self.sslopts["server_hostname"] = self._custom_endpoint or host
         self._read_buffer = BytesIO()
-        super(SSLTransport, self).__init__(host, port=port, socket_timeout=socket_timeout, **kwargs)
+        super(SSLTransport, self).__init__(
+            self._custom_endpoint or host, port=self._custom_port or port, socket_timeout=socket_timeout, **kwargs
+        )
 
     def _setup_transport(self):
         """Wrap the socket in an SSL object."""

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py

@@ -122,6 +122,7 @@ class AMQPClientAsync(AMQPClientSync):
     :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
      the Event Hubs service, allowing network requests to be routed through any application gateways or
      other paths needed for the host environment. Default is None.
+     Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
      If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
     :paramtype custom_endpoint_address: str
     :keyword connection_verify: Path to the custom CA_BUNDLE file of the SSL certificate which is used to
@@ -478,6 +479,7 @@ class SendClientAsync(SendClientSync, AMQPClientAsync):
     :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
      the Event Hubs service, allowing network requests to be routed through any application gateways or
      other paths needed for the host environment. Default is None.
+     Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
      If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
     :paramtype custom_endpoint_address: str
     :keyword connection_verify: Path to the custom CA_BUNDLE file of the SSL certificate which is used to
@@ -686,6 +688,7 @@ class ReceiveClientAsync(ReceiveClientSync, AMQPClientAsync):
     :keyword custom_endpoint_address: The custom endpoint address to use for establishing a connection to
      the Event Hubs service, allowing network requests to be routed through any application gateways or
      other paths needed for the host environment. Default is None.
+     Unless specified otherwise, default transport type is TransportType.AmqpOverWebsockets.
      If port is not specified in the `custom_endpoint_address`, by default port 443 will be used.
     :paramtype custom_endpoint_address: str
     :keyword connection_verify: Path to the custom CA_BUNDLE file of the SSL certificate which is used to

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py

@@ -115,10 +115,15 @@ def __init__(  # pylint:disable=too-many-locals,too-many-statements
         # Custom Endpoint
         custom_endpoint_address = kwargs.get("custom_endpoint_address")
         custom_endpoint = None
+        custom_port = None
         if custom_endpoint_address:
             custom_parsed_url = urlparse(custom_endpoint_address)
-            custom_port = custom_parsed_url.port or WEBSOCKET_PORT
-            custom_endpoint = f"{custom_parsed_url.hostname}:{custom_port}{custom_parsed_url.path}"
+            if transport_type.value == TransportType.Amqp.value:
+                custom_port = custom_parsed_url.port or SECURE_PORT
+                custom_endpoint = f"{custom_parsed_url.hostname}"
+            else:
+                custom_port = custom_parsed_url.port or WEBSOCKET_PORT
+                custom_endpoint = f"{custom_parsed_url.hostname}:{custom_port}{custom_parsed_url.path}"
         self._container_id: str = container_id or str(uuid.uuid4())
         self._network_trace = network_trace
         self._network_trace_params = {"amqpConnection": self._container_id, "amqpSession": "", "amqpLink": ""}
@@ -145,6 +150,7 @@ def __init__(  # pylint:disable=too-many-locals,too-many-statements
                 credential=kwargs["sasl_credential"],
                 port=self._port,
                 custom_endpoint=custom_endpoint,
+                custom_port=custom_port,
                 socket_timeout=self._socket_timeout,
                 network_trace_params=self._network_trace_params,
                 **kwargs,