pulsar.asyncio.Consumer

class documentation

class Consumer: (source)

Constructor: Consumer(consumer, schema)

View In Hierarchy

The Pulsar message consumer, used to subscribe to messages from a topic.

Method __init__ Create the consumer. Users should not call this constructor directly. Instead, create the consumer via Client.subscribe.
Async Method acknowledge Acknowledge the reception of a single message asynchronously.
Async Method acknowledge_cumulative Acknowledge the reception of all the messages in the stream up to (and including) the provided message asynchronously.
Async Method close Close the consumer asynchronously.
Method consumer_name Return the consumer name.
Async Method get_last_message_id Asynchronously get the last message id.
Async Method negative_acknowledge Acknowledge the failure to process a single message asynchronously.
Async Method receive Receive a single message asynchronously.
Method redeliver_unacknowledged_messages 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...
Async Method seek Reset the subscription associated with this consumer to a specific message id or publish timestamp asynchronously.
Method subscription_name Return the subscription name.
Method topic Return the topic this consumer is subscribed to.
Async Method unsubscribe Unsubscribe the current consumer from the topic asynchronously.
Instance Variable _consumer Undocumented
Instance Variable _schema Undocumented
def __init__(self, consumer: _pulsar.Consumer, schema: pulsar.schema.Schema): (source)

Create the consumer. Users should not call this constructor directly. Instead, create the consumer via Client.subscribe.

Parameters
consumer:_pulsar.ConsumerThe underlying Consumer object from the C extension.
schema:pulsar.schema.SchemaThe schema of the data that will be received by this consumer.
async def acknowledge(self, message: pulsar.Message | pulsar.MessageId | _pulsar.Message | _pulsar.MessageId): (source)

Acknowledge the reception of a single message asynchronously.

Parameters
message:Message, MessageId, _pulsar.Message, _pulsar.MessageIdThe received message or message id.
Raises
PulsarException
async def acknowledge_cumulative(self, message: pulsar.Message | pulsar.MessageId | _pulsar.Message | _pulsar.MessageId): (source)

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.MessageIdThe received message or message id.
Raises
PulsarException
async def close(self): (source)

Close the consumer asynchronously.

Raises
PulsarException
def consumer_name(self) -> str: (source)

Return the consumer name.

async def get_last_message_id(self) -> _pulsar.MessageId: (source)

Asynchronously get the last message id.

async def negative_acknowledge(self, message: pulsar.Message | pulsar.MessageId | _pulsar.Message | _pulsar.MessageId): (source)

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:pulsar.Message | pulsar.MessageId | _pulsar.Message | _pulsar.MessageIdThe received message or message id.
async def receive(self) -> pulsar.Message: (source)

Receive a single message asynchronously.

Returns
pulsar.MessageThe message received.
Raises
PulsarException
def redeliver_unacknowledged_messages(self): (source)

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.

async def seek(self, messageid: pulsar.MessageId | int): (source)

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 intThe message id for seek, OR an integer event time (timestamp) to seek to
Raises
PulsarException
def subscription_name(self) -> str: (source)

Return the subscription name.

def topic(self) -> str: (source)

Return the topic this consumer is subscribed to.

async def unsubscribe(self): (source)

Unsubscribe the current consumer from the topic asynchronously.

Raises
PulsarException
_consumer = (source)

Undocumented

_schema = (source)

Undocumented