Skip to content

Commit a189711

Browse files
authored
chore: add async example (#291)
1 parent 8cfeac6 commit a189711

File tree

1 file changed

+71
-0
lines changed

1 file changed

+71
-0
lines changed

examples/async.py

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
#!/usr/bin/env python3
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one
4+
# or more contributor license agreements. See the NOTICE file
5+
# distributed with this work for additional information
6+
# regarding copyright ownership. The ASF licenses this file
7+
# to you under the Apache License, Version 2.0 (the
8+
# "License"); you may not use this file except in compliance
9+
# with the License. You may obtain a copy of the License at
10+
#
11+
# http://www.apache.org/licenses/LICENSE-2.0
12+
#
13+
# Unless required by applicable law or agreed to in writing,
14+
# software distributed under the License is distributed on an
15+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
# KIND, either express or implied. See the License for the
17+
# specific language governing permissions and limitations
18+
# under the License.
19+
#
20+
21+
# NOTE: requires version 3.9.0+ of pulsar-client library
22+
import asyncio
23+
from pulsar.asyncio import Client, Producer, Consumer
24+
from pulsar import BatchingType
25+
26+
NUM_MESSAGES = 100
27+
TOPIC_NAME = 'my-async-topic'
28+
SUBSCRIPTION_NAME = 'my-async-subscription'
29+
SERVICE_URL = 'pulsar://localhost:6650'
30+
31+
async def produce(producer: Producer, id: int) -> None:
32+
await producer.send((f'hello-{id}').encode('utf-8'), None)
33+
await producer.flush()
34+
35+
async def consume(consumer: Consumer) -> None:
36+
msg = await consumer.receive()
37+
print("Received message '{0}' id='{1}'".format(msg.data().decode('utf-8'), msg.message_id()))
38+
await consumer.acknowledge(msg)
39+
40+
async def main() -> None:
41+
client: Client = Client(SERVICE_URL)
42+
consumer = await client.subscribe(TOPIC_NAME, SUBSCRIPTION_NAME,
43+
properties={
44+
"consumer-name": "test-consumer-name",
45+
"consumer-id": "test-consumer-id"
46+
})
47+
48+
producer = await client.create_producer(
49+
TOPIC_NAME,
50+
block_if_queue_full=True,
51+
batching_enabled=True,
52+
batching_max_publish_delay_ms=10,
53+
properties={
54+
"producer-name": "test-producer-name",
55+
"producer-id": "test-producer-id"
56+
},
57+
batching_type=BatchingType.KeyBased
58+
)
59+
60+
tasks = []
61+
for id in range(NUM_MESSAGES):
62+
tasks.append(asyncio.create_task(produce(producer, id)))
63+
tasks.append(asyncio.create_task(consume(consumer)))
64+
await asyncio.gather(*tasks)
65+
66+
await producer.close()
67+
await consumer.close()
68+
await client.close()
69+
70+
if __name__ == '__main__':
71+
asyncio.run(main())

0 commit comments

Comments
 (0)