Skip to content
This repository was archived by the owner on Mar 9, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion google/cloud/pubsub_v1/publisher/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,16 @@ def __init__(self, batch_settings=(), publisher_options=(), **kwargs):
target=os.environ.get("PUBSUB_EMULATOR_HOST")
)

client_options = kwargs.pop("client_options", None)
if (
client_options
and "api_endpoint" in client_options
and isinstance(client_options["api_endpoint"], six.string_types)
):
self._target = client_options["api_endpoint"]
else:
self._target = publisher_client.PublisherClient.SERVICE_ADDRESS

# Use a custom channel.
# We need this in order to set appropriate default message size and
# keepalive options.
Expand Down Expand Up @@ -217,7 +227,7 @@ def target(self):
Returns:
str: The location of the API.
"""
return publisher_client.PublisherClient.SERVICE_ADDRESS
return self._target

def _get_or_create_sequencer(self, topic, ordering_key):
""" Get an existing sequencer or create a new one given the (topic,
Expand Down
14 changes: 13 additions & 1 deletion google/cloud/pubsub_v1/subscriber/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

import os
import pkg_resources
import six

import grpc

Expand Down Expand Up @@ -79,6 +80,17 @@ def __init__(self, **kwargs):
target=os.environ.get("PUBSUB_EMULATOR_HOST")
)

# api_endpoint wont be applied if 'transport' is passed in.
client_options = kwargs.pop("client_options", None)
if (
client_options
and "api_endpoint" in client_options
and isinstance(client_options["api_endpoint"], six.string_types)
):
self._target = client_options["api_endpoint"]
else:
self._target = subscriber_client.SubscriberClient.SERVICE_ADDRESS

# Use a custom channel.
# We need this in order to set appropriate default message size and
# keepalive options.
Expand Down Expand Up @@ -133,7 +145,7 @@ def target(self):
Returns:
str: The location of the API.
"""
return subscriber_client.SubscriberClient.SERVICE_ADDRESS
return self._target

@property
def api(self):
Expand Down
29 changes: 29 additions & 0 deletions tests/unit/pubsub_v1/publisher/test_publisher_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,35 @@ def test_init_w_custom_transport():
assert client.batch_settings.max_messages == 100


def test_init_w_api_endpoint():
client_options = {"api_endpoint": "testendpoint.google.com"}
client = publisher.Client(client_options=client_options)

assert isinstance(client.api, publisher_client.PublisherClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == "testendpoint.google.com"


def test_init_w_unicode_api_endpoint():
client_options = {"api_endpoint": u"testendpoint.google.com"}
Comment thread
pradn marked this conversation as resolved.
client = publisher.Client(client_options=client_options)

assert isinstance(client.api, publisher_client.PublisherClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == "testendpoint.google.com"


def test_init_w_empty_client_options():
client = publisher.Client(client_options={})

assert isinstance(client.api, publisher_client.PublisherClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == publisher_client.PublisherClient.SERVICE_ADDRESS


def test_init_emulator(monkeypatch):
monkeypatch.setenv("PUBSUB_EMULATOR_HOST", "/foo/bar/")
# NOTE: When the emulator host is set, a custom channel will be used, so
Expand Down
29 changes: 29 additions & 0 deletions tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,35 @@ def test_init_w_custom_transport():
assert client.api.transport is transport


def test_init_w_api_endpoint():
client_options = {"api_endpoint": "testendpoint.google.com"}
client = subscriber.Client(client_options=client_options)

assert isinstance(client.api, subscriber_client.SubscriberClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == "testendpoint.google.com"


def test_init_w_unicode_api_endpoint():
client_options = {"api_endpoint": u"testendpoint.google.com"}
client = subscriber.Client(client_options=client_options)

assert isinstance(client.api, subscriber_client.SubscriberClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == "testendpoint.google.com"


def test_init_w_empty_client_options():
client = subscriber.Client(client_options={})

assert isinstance(client.api, subscriber_client.SubscriberClient)
assert (client.api.transport._channel._channel.target()).decode(
"utf-8"
) == subscriber_client.SubscriberClient.SERVICE_ADDRESS


def test_init_emulator(monkeypatch):
monkeypatch.setenv("PUBSUB_EMULATOR_HOST", "/baz/bacon/")
# NOTE: When the emulator host is set, a custom channel will be used, so
Expand Down