| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| # pylint: disable=no-name-in-module,c-extension-no-member,protected-access |
| |
| """ |
| The Pulsar Python client APIs that work with the asyncio module. |
| """ |
| |
| import asyncio |
| import functools |
| from datetime import timedelta |
| from typing import Any, Callable, List, Union |
| |
| import _pulsar |
| from _pulsar import ( |
| InitialPosition, |
| CompressionType, |
| PartitionsRoutingMode, |
| BatchingType, |
| ProducerAccessMode, |
| RegexSubscriptionMode, |
| ConsumerCryptoFailureAction, |
| ) |
| import pulsar |
| from pulsar import _check_type |
| |
| class PulsarException(BaseException): |
| """ |
| The exception that wraps the Pulsar error code |
| """ |
| |
| def __init__(self, result: pulsar.Result) -> None: |
| """ |
| Create the Pulsar exception. |
| |
| Parameters |
| ---------- |
| result: pulsar.Result |
| The error code of the underlying Pulsar APIs. |
| """ |
| self._result = result |
| |
| def error(self) -> pulsar.Result: |
| """ |
| Returns the Pulsar error code. |
| """ |
| return self._result |
| |
| def __str__(self): |
| """ |
| Convert the exception to string. |
| """ |
| return f'{self._result.value} {self._result.name}' |
| |
| class Producer: |
| """ |
| The Pulsar message producer, used to publish messages on a topic. |
| """ |
| |
| def __init__(self, producer: _pulsar.Producer, schema: pulsar.schema.Schema) -> None: |
| """ |
| Create the producer. |
| Users should not call this constructor directly. Instead, create the |
| producer via `Client.create_producer`. |
| |
| Parameters |
| ---------- |
| producer: _pulsar.Producer |
| The underlying Producer object from the C extension. |
| schema: pulsar.schema.Schema |
| The schema of the data that will be sent by this producer. |
| """ |
| self._producer = producer |
| self._schema = schema |
| |
| # pylint: disable=too-many-arguments,too-many-locals,too-many-positional-arguments |
| async def send(self, content: Any, |
| properties: dict | None = None, |
| partition_key: str | None = None, |
| ordering_key: str | None = None, |
| sequence_id: int | None = None, |
| replication_clusters: List[str] | None = None, |
| disable_replication: bool | None = None, |
| event_timestamp: int | None = None, |
| deliver_at: int | None = None, |
| deliver_after: timedelta | None = None) -> pulsar.MessageId: |
| """ |
| Send a message asynchronously. |
| |
| parameters |
| ---------- |
| content: Any |
| The message payload, whose type should respect the schema defined in |
| `Client.create_producer`. |
| properties: dict | None |
| A dict of application-defined string properties. |
| partition_key: str | None |
| Sets the partition key for the message routing. A hash of this key is |
| used to determine the message's topic partition. |
| ordering_key: str | None |
| Sets the ordering key for the message routing. |
| sequence_id: int | None |
| Specify a custom sequence id for the message being published. |
| replication_clusters: List[str] | None |
| Override namespace replication clusters. Note that it is the caller's responsibility |
| to provide valid cluster names and that all clusters have been previously configured |
| as topics. Given an empty list, the message will replicate per the namespace |
| configuration. |
| disable_replication: bool | None |
| Do not replicate this message. |
| event_timestamp: int | None |
| Timestamp in millis of the timestamp of event creation |
| deliver_at: int | None |
| Specify the message should not be delivered earlier than the specified timestamp. |
| deliver_after: timedelta | None |
| Specify a delay in timedelta for the delivery of the messages. |
| |
| Returns |
| ------- |
| pulsar.MessageId |
| The message id that represents the persisted position of the message. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| builder = _pulsar.MessageBuilder() |
| builder.content(self._schema.encode(content)) |
| |
| if properties is not None: |
| for k, v in properties.items(): |
| builder.property(k, v) |
| if partition_key is not None: |
| builder.partition_key(partition_key) |
| if ordering_key is not None: |
| builder.ordering_key(ordering_key) |
| if sequence_id is not None: |
| builder.sequence_id(sequence_id) |
| if replication_clusters is not None: |
| builder.replication_clusters(replication_clusters) |
| if disable_replication is not None: |
| builder.disable_replication(disable_replication) |
| if event_timestamp is not None: |
| builder.event_timestamp(event_timestamp) |
| if deliver_at is not None: |
| builder.deliver_at(deliver_at) |
| if deliver_after is not None: |
| builder.deliver_after(deliver_after) |
| |
| future = asyncio.get_running_loop().create_future() |
| self._producer.send_async(builder.build(), functools.partial(_set_future, future)) |
| msg_id = await future |
| return pulsar.MessageId( |
| msg_id.partition(), |
| msg_id.ledger_id(), |
| msg_id.entry_id(), |
| msg_id.batch_index(), |
| ) |
| |
| async def flush(self) -> None: |
| """ |
| Flush all the messages buffered in the producer asynchronously. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._producer.flush_async(functools.partial(_set_future, future, value=None)) |
| await future |
| |
| async def close(self) -> None: |
| """ |
| Close the producer. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._producer.close_async(functools.partial(_set_future, future, value=None)) |
| await future |
| |
| def topic(self): |
| """ |
| Return the topic which producer is publishing to |
| """ |
| return self._producer.topic() |
| |
| def producer_name(self): |
| """ |
| Return the producer name which could have been assigned by the |
| system or specified by the client |
| """ |
| return self._producer.producer_name() |
| |
| def last_sequence_id(self): |
| """ |
| Return the last sequence id that was published and acknowledged by this producer. |
| |
| The sequence id can be either automatically assigned or custom set on the message. |
| After recreating a producer with the same name, this will return the sequence id |
| of the last message that was published in the previous session, or -1 if no |
| message was ever published. |
| """ |
| return self._producer.last_sequence_id() |
| |
| def is_connected(self) -> bool: |
| """ |
| Check if the producer is connected or not. |
| """ |
| return self._producer.is_connected() |
| |
| class Consumer: |
| """ |
| The Pulsar message consumer, used to subscribe to messages from a topic. |
| """ |
| |
| def __init__(self, consumer: _pulsar.Consumer, schema: pulsar.schema.Schema) -> None: |
| """ |
| Create the consumer. |
| Users should not call this constructor directly. Instead, create the |
| consumer via `Client.subscribe`. |
| |
| Parameters |
| ---------- |
| consumer: _pulsar.Consumer |
| The underlying Consumer object from the C extension. |
| schema: pulsar.schema.Schema |
| The schema of the data that will be received by this consumer. |
| """ |
| self._consumer = consumer |
| self._schema = schema |
| |
| async def receive(self) -> pulsar.Message: |
| """ |
| Receive a single message asynchronously. |
| |
| Returns |
| ------- |
| pulsar.Message |
| The message received. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._consumer.receive_async(functools.partial(_set_future, future)) |
| msg = await future |
| m = pulsar.Message() |
| m._message = msg |
| m._schema = self._schema |
| return m |
| |
| async def acknowledge( |
| self, |
| message: Union[pulsar.Message, pulsar.MessageId, |
| _pulsar.Message, _pulsar.MessageId] |
| ) -> None: |
| """ |
| Acknowledge the reception of a single message asynchronously. |
| |
| Parameters |
| ---------- |
| message : Message, MessageId, _pulsar.Message, _pulsar.MessageId |
| The received message or message id. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| if isinstance(message, pulsar.Message): |
| msg = message._message |
| elif isinstance(message, pulsar.MessageId): |
| msg = message._msg_id |
| else: |
| msg = message |
| self._consumer.acknowledge_async(msg, functools.partial(_set_future, future, value=None)) |
| await future |
| |
| async def acknowledge_cumulative( |
| self, |
| message: Union[pulsar.Message, pulsar.MessageId, |
| _pulsar.Message, _pulsar.MessageId] |
| ) -> None: |
| """ |
| Acknowledge the reception of all the messages in the stream up to (and |
| including) the provided message asynchronously. |
| |
| Parameters |
| ---------- |
| message : Message, MessageId, _pulsar.Message, _pulsar.MessageId |
| The received message or message id. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| if isinstance(message, pulsar.Message): |
| msg = message._message |
| elif isinstance(message, pulsar.MessageId): |
| msg = message._msg_id |
| else: |
| msg = message |
| self._consumer.acknowledge_cumulative_async( |
| msg, functools.partial(_set_future, future, value=None) |
| ) |
| await future |
| |
| async def negative_acknowledge( |
| self, |
| message: Union[pulsar.Message, pulsar.MessageId, _pulsar.Message, _pulsar.MessageId] |
| ) -> None: |
| """ |
| Acknowledge the failure to process a single message asynchronously. |
| |
| When a message is "negatively acked" it will be marked for redelivery after |
| some fixed delay. The delay is configurable when constructing the consumer |
| with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}. |
| This call is not blocking. |
| |
| Parameters |
| ---------- |
| message: |
| The received message or message id. |
| """ |
| if isinstance(message, pulsar.Message): |
| msg = message._message |
| elif isinstance(message, pulsar.MessageId): |
| msg = message._msg_id |
| else: |
| msg = message |
| await asyncio.to_thread(self._consumer.negative_acknowledge, msg) |
| |
| async def unsubscribe(self) -> None: |
| """ |
| Unsubscribe the current consumer from the topic asynchronously. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._consumer.unsubscribe_async(functools.partial(_set_future, future, value=None)) |
| await future |
| |
| async def seek(self, messageid: Union[pulsar.MessageId, int]) -> None: |
| """ |
| Reset the subscription associated with this consumer to a specific |
| message id or publish timestamp asynchronously. |
| |
| The message id can either be a specific message or represent the first |
| or last messages in the topic. |
| |
| Parameters |
| ---------- |
| messageid : MessageId or int |
| The message id for seek, OR an integer event time (timestamp) to |
| seek to |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| if isinstance(messageid, pulsar.MessageId): |
| msg_id = messageid._msg_id |
| elif isinstance(messageid, int): |
| msg_id = messageid |
| else: |
| raise ValueError(f"invalid messageid type {type(messageid)}") |
| self._consumer.seek_async( |
| msg_id, functools.partial(_set_future, future, value=None) |
| ) |
| await future |
| |
| async def close(self) -> None: |
| """ |
| Close the consumer asynchronously. |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._consumer.close_async(functools.partial(_set_future, future, value=None)) |
| await future |
| |
| async def get_last_message_id(self) -> _pulsar.MessageId: |
| """ |
| Asynchronously get the last message id. |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._consumer.get_last_message_id_async(functools.partial(_set_future, future)) |
| id = await future |
| return id |
| |
| def redeliver_unacknowledged_messages(self): |
| """ |
| Redelivers all the unacknowledged messages. In failover mode, the |
| request is ignored if the consumer is not active for the given topic. In |
| shared mode, the consumer's messages to be redelivered are distributed |
| across all the connected consumers. This is a non-blocking call and |
| doesn't throw an exception. In case the connection breaks, the messages |
| are redelivered after reconnect. |
| """ |
| self._consumer.redeliver_unacknowledged_messages() |
| |
| def topic(self) -> str: |
| """ |
| Return the topic this consumer is subscribed to. |
| """ |
| return self._consumer.topic() |
| |
| def subscription_name(self) -> str: |
| """ |
| Return the subscription name. |
| """ |
| return self._consumer.subscription_name() |
| |
| def consumer_name(self) -> str: |
| """ |
| Return the consumer name. |
| """ |
| return self._consumer.consumer_name() |
| |
| class Client: |
| """ |
| The asynchronous version of `pulsar.Client`. |
| """ |
| |
| def __init__(self, service_url, **kwargs) -> None: |
| """ |
| See `pulsar.Client.__init__` |
| """ |
| self._client: _pulsar.Client = pulsar.Client(service_url, **kwargs)._client |
| |
| # pylint: disable=too-many-arguments,too-many-locals,too-many-positional-arguments |
| async def create_producer(self, topic: str, |
| producer_name: str | None = None, |
| schema: pulsar.schema.Schema | None = None, |
| initial_sequence_id: int | None = None, |
| send_timeout_millis: int = 30000, |
| compression_type: CompressionType = CompressionType.NONE, |
| max_pending_messages: int = 1000, |
| max_pending_messages_across_partitions: int = 50000, |
| block_if_queue_full: bool = False, |
| batching_enabled: bool = True, |
| batching_max_messages: int = 1000, |
| batching_max_allowed_size_in_bytes: int = 128*1024, |
| batching_max_publish_delay_ms: int = 10, |
| chunking_enabled: bool = False, |
| message_routing_mode: PartitionsRoutingMode = |
| PartitionsRoutingMode.RoundRobinDistribution, |
| lazy_start_partitioned_producers: bool = False, |
| properties: dict | None = None, |
| batching_type: BatchingType = BatchingType.Default, |
| encryption_key: str | None = None, |
| crypto_key_reader: pulsar.CryptoKeyReader | None = None, |
| access_mode: ProducerAccessMode = ProducerAccessMode.Shared, |
| message_router: Callable[[pulsar.Message, int], int] | None = None, |
| ) -> Producer: |
| """ |
| Create a new producer on a given topic |
| |
| Parameters |
| ---------- |
| topic: str |
| The topic name |
| producer_name: str | None |
| Specify a name for the producer. If not assigned, the system will generate a globally |
| unique name which can be accessed with `Producer.producer_name()`. When specifying a |
| name, it is up to the user to ensure that, for a given topic, the producer name is |
| unique across all Pulsar's clusters. |
| schema: pulsar.schema.Schema | None, default=None |
| Define the schema of the data that will be published by this producer. |
| initial_sequence_id: int | None, default=None |
| Set the baseline for the sequence ids for messages published by |
| the producer. |
| send_timeout_millis: int, default=30000 |
| If a message is not acknowledged by the server before the |
| send_timeout expires, an error will be reported. |
| compression_type: CompressionType, default=CompressionType.NONE |
| Set the compression type for the producer. |
| max_pending_messages: int, default=1000 |
| Set the max size of the queue holding the messages pending to |
| receive an acknowledgment from the broker. |
| max_pending_messages_across_partitions: int, default=50000 |
| Set the max size of the queue holding the messages pending to |
| receive an acknowledgment across partitions. |
| block_if_queue_full: bool, default=False |
| Set whether send operations should block when the outgoing |
| message queue is full. |
| batching_enabled: bool, default=True |
| Enable automatic message batching. Note that, unlike the synchronous producer API in |
| ``pulsar.Client.create_producer``, batching is enabled by default for the asyncio |
| producer. |
| batching_max_messages: int, default=1000 |
| Maximum number of messages in a batch. |
| batching_max_allowed_size_in_bytes: int, default=128*1024 |
| Maximum size in bytes of a batch. |
| batching_max_publish_delay_ms: int, default=10 |
| The batch interval in milliseconds. |
| chunking_enabled: bool, default=False |
| Enable chunking of large messages. |
| message_routing_mode: PartitionsRoutingMode, |
| default=PartitionsRoutingMode.RoundRobinDistribution |
| Set the message routing mode for the partitioned producer. |
| lazy_start_partitioned_producers: bool, default=False |
| Start partitioned producers lazily on demand. |
| properties: dict | None, default=None |
| Sets the properties for the producer. |
| batching_type: BatchingType, default=BatchingType.Default |
| Sets the batching type for the producer. |
| encryption_key: str | None, default=None |
| The key used for symmetric encryption. |
| crypto_key_reader: pulsar.CryptoKeyReader | None, default=None |
| Symmetric encryption class implementation. |
| access_mode: ProducerAccessMode, default=ProducerAccessMode.Shared |
| Set the type of access mode that the producer requires on the topic. |
| message_router: Callable[[pulsar.Message, int], int] | None, default=None |
| A custom message router function that takes a Message and the |
| number of partitions and returns the partition index. |
| |
| Returns |
| ------- |
| Producer |
| The producer created |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| if batching_enabled and chunking_enabled: |
| raise ValueError("Batching and chunking of messages can't be enabled together.") |
| |
| if schema is None: |
| schema = pulsar.schema.BytesSchema() |
| schema.attach_client(self._client) |
| |
| future = asyncio.get_running_loop().create_future() |
| conf = _pulsar.ProducerConfiguration() |
| if producer_name is not None: |
| conf.producer_name(producer_name) |
| conf.schema(schema.schema_info()) |
| if initial_sequence_id is not None: |
| conf.initial_sequence_id(initial_sequence_id) |
| conf.send_timeout_millis(send_timeout_millis) |
| conf.compression_type(compression_type) |
| conf.max_pending_messages(max_pending_messages) |
| conf.max_pending_messages_across_partitions(max_pending_messages_across_partitions) |
| conf.block_if_queue_full(block_if_queue_full) |
| conf.batching_enabled(batching_enabled) |
| conf.batching_max_messages(batching_max_messages) |
| conf.batching_max_allowed_size_in_bytes(batching_max_allowed_size_in_bytes) |
| conf.batching_max_publish_delay_ms(batching_max_publish_delay_ms) |
| conf.chunking_enabled(chunking_enabled) |
| conf.partitions_routing_mode(message_routing_mode) |
| conf.lazy_start_partitioned_producers(lazy_start_partitioned_producers) |
| if properties is not None: |
| for k, v in properties.items(): |
| conf.property(k, v) |
| conf.batching_type(batching_type) |
| if encryption_key is not None: |
| conf.encryption_key(encryption_key) |
| if crypto_key_reader is not None: |
| conf.crypto_key_reader(crypto_key_reader.cryptoKeyReader) |
| conf.access_mode(access_mode) |
| if message_router is not None: |
| def underlying_router(msg: _pulsar.Message, num_partitions: int) -> int: |
| return message_router(pulsar.Message._wrap(msg), num_partitions) |
| conf.message_router(underlying_router) |
| |
| self._client.create_producer_async( |
| topic, conf, functools.partial(_set_future, future) |
| ) |
| return Producer(await future, schema) |
| |
| # pylint: disable=too-many-arguments,too-many-locals,too-many-branches,too-many-positional-arguments |
| async def subscribe(self, topic: Union[str, List[str]], |
| subscription_name: str, |
| consumer_type: pulsar.ConsumerType = |
| pulsar.ConsumerType.Exclusive, |
| schema: pulsar.schema.Schema | None = None, |
| receiver_queue_size: int = 1000, |
| max_total_receiver_queue_size_across_partitions: int = |
| 50000, |
| consumer_name: str | None = None, |
| unacked_messages_timeout_ms: int | None = None, |
| broker_consumer_stats_cache_time_ms: int = 30000, |
| negative_ack_redelivery_delay_ms: int = 60000, |
| is_read_compacted: bool = False, |
| properties: dict | None = None, |
| initial_position: InitialPosition = InitialPosition.Latest, |
| crypto_key_reader: pulsar.CryptoKeyReader | None = None, |
| replicate_subscription_state_enabled: bool = False, |
| max_pending_chunked_message: int = 10, |
| auto_ack_oldest_chunked_message_on_queue_full: bool = False, |
| start_message_id_inclusive: bool = False, |
| batch_receive_policy: pulsar.ConsumerBatchReceivePolicy | None = |
| None, |
| key_shared_policy: pulsar.ConsumerKeySharedPolicy | None = |
| None, |
| batch_index_ack_enabled: bool = False, |
| regex_subscription_mode: RegexSubscriptionMode = |
| RegexSubscriptionMode.PersistentOnly, |
| dead_letter_policy: pulsar.ConsumerDeadLetterPolicy | None = |
| None, |
| crypto_failure_action: ConsumerCryptoFailureAction = |
| ConsumerCryptoFailureAction.FAIL, |
| is_pattern_topic: bool = False) -> Consumer: |
| """ |
| Subscribe to the given topic and subscription combination. |
| |
| Parameters |
| ---------- |
| topic: str, List[str], or regex pattern |
| The name of the topic, list of topics or regex pattern. |
| When `is_pattern_topic` is True, `topic` is treated as a regex. |
| subscription_name: str |
| The name of the subscription. |
| consumer_type: pulsar.ConsumerType, default=pulsar.ConsumerType.Exclusive |
| Select the subscription type to be used when subscribing to the topic. |
| schema: pulsar.schema.Schema | None, default=None |
| Define the schema of the data that will be received by this consumer. |
| receiver_queue_size: int, default=1000 |
| Sets the size of the consumer receive queue. |
| max_total_receiver_queue_size_across_partitions: int, default=50000 |
| Set the max total receiver queue size across partitions. |
| consumer_name: str | None, default=None |
| Sets the consumer name. |
| unacked_messages_timeout_ms: int | None, default=None |
| Sets the timeout in milliseconds for unacknowledged messages. |
| broker_consumer_stats_cache_time_ms: int, default=30000 |
| Sets the time duration for which the broker-side consumer stats |
| will be cached in the client. |
| negative_ack_redelivery_delay_ms: int, default=60000 |
| The delay after which to redeliver the messages that failed to be |
| processed. |
| is_read_compacted: bool, default=False |
| Selects whether to read the compacted version of the topic. |
| properties: dict | None, default=None |
| Sets the properties for the consumer. |
| initial_position: InitialPosition, default=InitialPosition.Latest |
| Set the initial position of a consumer when subscribing to the topic. |
| crypto_key_reader: pulsar.CryptoKeyReader | None, default=None |
| Symmetric encryption class implementation. |
| replicate_subscription_state_enabled: bool, default=False |
| Set whether the subscription status should be replicated. |
| max_pending_chunked_message: int, default=10 |
| Consumer buffers chunk messages into memory until it receives all the chunks. |
| auto_ack_oldest_chunked_message_on_queue_full: bool, default=False |
| Automatically acknowledge oldest chunked messages on queue |
| full. |
| start_message_id_inclusive: bool, default=False |
| Set the consumer to include the given position of any reset |
| operation. |
| batch_receive_policy: pulsar.ConsumerBatchReceivePolicy | None, default=None |
| Set the batch collection policy for batch receiving. |
| key_shared_policy: pulsar.ConsumerKeySharedPolicy | None, default=None |
| Set the key shared policy for use when the ConsumerType is |
| KeyShared. |
| batch_index_ack_enabled: bool, default=False |
| Enable the batch index acknowledgement. |
| regex_subscription_mode: RegexSubscriptionMode, |
| default=RegexSubscriptionMode.PersistentOnly |
| Set the regex subscription mode for use when the topic is a regex |
| pattern. |
| dead_letter_policy: pulsar.ConsumerDeadLetterPolicy | None, default=None |
| Set dead letter policy for consumer. |
| crypto_failure_action: ConsumerCryptoFailureAction, |
| default=ConsumerCryptoFailureAction.FAIL |
| Set the behavior when the decryption fails. |
| is_pattern_topic: bool, default=False |
| Whether `topic` is a regex pattern. If it's True when `topic` is a list, a ValueError |
| will be raised. |
| |
| Returns |
| ------- |
| Consumer |
| The consumer created |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| if schema is None: |
| schema = pulsar.schema.BytesSchema() |
| |
| future = asyncio.get_running_loop().create_future() |
| conf = _pulsar.ConsumerConfiguration() |
| conf.consumer_type(consumer_type) |
| conf.regex_subscription_mode(regex_subscription_mode) |
| conf.read_compacted(is_read_compacted) |
| conf.receiver_queue_size(receiver_queue_size) |
| conf.max_total_receiver_queue_size_across_partitions( |
| max_total_receiver_queue_size_across_partitions |
| ) |
| if consumer_name: |
| conf.consumer_name(consumer_name) |
| if unacked_messages_timeout_ms: |
| conf.unacked_messages_timeout_ms(unacked_messages_timeout_ms) |
| |
| conf.negative_ack_redelivery_delay_ms(negative_ack_redelivery_delay_ms) |
| conf.broker_consumer_stats_cache_time_ms(broker_consumer_stats_cache_time_ms) |
| if properties: |
| for k, v in properties.items(): |
| conf.property(k, v) |
| conf.subscription_initial_position(initial_position) |
| |
| conf.schema(schema.schema_info()) |
| |
| if crypto_key_reader: |
| conf.crypto_key_reader(crypto_key_reader.cryptoKeyReader) |
| |
| conf.replicate_subscription_state_enabled(replicate_subscription_state_enabled) |
| conf.max_pending_chunked_message(max_pending_chunked_message) |
| conf.auto_ack_oldest_chunked_message_on_queue_full( |
| auto_ack_oldest_chunked_message_on_queue_full |
| ) |
| conf.start_message_id_inclusive(start_message_id_inclusive) |
| if batch_receive_policy: |
| conf.batch_receive_policy(batch_receive_policy.policy()) |
| |
| if key_shared_policy: |
| conf.key_shared_policy(key_shared_policy.policy()) |
| conf.batch_index_ack_enabled(batch_index_ack_enabled) |
| if dead_letter_policy: |
| conf.dead_letter_policy(dead_letter_policy.policy()) |
| conf.crypto_failure_action(crypto_failure_action) |
| |
| if isinstance(topic, str): |
| if is_pattern_topic: |
| self._client.subscribe_async_pattern( |
| topic, subscription_name, conf, |
| functools.partial(_set_future, future) |
| ) |
| else: |
| self._client.subscribe_async( |
| topic, subscription_name, conf, |
| functools.partial(_set_future, future) |
| ) |
| elif isinstance(topic, list): |
| if is_pattern_topic: |
| raise ValueError( |
| "Argument 'topic' must be a string when " |
| "'is_pattern_topic' is True; lists of topics do not " |
| "support pattern subscriptions" |
| ) |
| self._client.subscribe_async_topics( |
| topic, subscription_name, conf, |
| functools.partial(_set_future, future) |
| ) |
| else: |
| raise ValueError( "Argument 'topic' is expected to be of type 'str' or 'list'") |
| |
| schema.attach_client(self._client) |
| return Consumer(await future, schema) |
| |
| def shutdown(self) -> None: |
| """ |
| Shutdown the client and all the associated producers and consumers |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| self._client.shutdown() |
| |
| async def get_topic_partitions(self, topic: str) -> List[str]: |
| """ |
| Get the list of partitions for a given topic in asynchronous mode. |
| |
| If the topic is partitioned, this will return a list of partition names. If the topic is not partitioned, the returned list will contain the topic name itself. |
| |
| This can be used to discover the partitions and create Reader, Consumer or Producer instances directly on a particular partition. |
| |
| Parameters |
| ---------- |
| |
| topic: str |
| the topic name to lookup |
| |
| Returns |
| ------- |
| list |
| a list of partition names |
| """ |
| _check_type(str, topic, 'topic') |
| future = asyncio.get_running_loop().create_future() |
| self._client.get_topic_partitions_async(topic, functools.partial(_set_future, future)) |
| id = await future |
| return id |
| |
| async def close(self) -> None: |
| """ |
| Close the client and all the associated producers and consumers |
| |
| Raises |
| ------ |
| PulsarException |
| """ |
| future = asyncio.get_running_loop().create_future() |
| self._client.close_async( |
| functools.partial(_set_future, future, value=None) |
| ) |
| await future |
| |
| def _set_future(future: asyncio.Future, result: _pulsar.Result, value: Any): |
| def complete(): |
| if result == _pulsar.Result.Ok: |
| future.set_result(value) |
| else: |
| future.set_exception(PulsarException(result)) |
| future.get_loop().call_soon_threadsafe(complete) |
