Skip to content

Commit b1c9487

Browse files
authored
feat: add producer connectivity functionality (#286)
1 parent d781707 commit b1c9487

File tree

2 files changed

+13
-0
lines changed

2 files changed

+13
-0
lines changed

pulsar/asyncio.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -220,6 +220,12 @@ def last_sequence_id(self):
220220
message was ever published.
221221
"""
222222
return self._producer.last_sequence_id()
223+
224+
def is_connected(self) -> bool:
225+
"""
226+
Check if the producer is connected or not.
227+
"""
228+
return self._producer.is_connected()
223229

224230
class Consumer:
225231
"""

tests/asyncio_test.py

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,13 @@ async def test_close_producer(self):
157157
except PulsarException as e:
158158
self.assertEqual(e.error(), pulsar.Result.AlreadyClosed)
159159

160+
async def test_producer_is_connected(self):
161+
topic = f'asyncio-test-producer-is-connected-{time.time()}'
162+
producer = await self._client.create_producer(topic)
163+
self.assertTrue(producer.is_connected())
164+
await producer.close()
165+
self.assertFalse(producer.is_connected())
166+
160167
async def _prepare_messages(self, producer: Producer) -> List[pulsar.MessageId]:
161168
msg_ids = []
162169
for i in range(5):

0 commit comments

Comments
 (0)