You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
importloggingimportasyncioimportnatsfromnats.js.apiimportConsumerConfig, StreamConfig, StorageTypeimportsyslogging.basicConfig(stream=sys.stdout, level=logging.INFO)
asyncdefconsume():
nc=awaitnats.connect()
js=nc.jetstream()
awaitjs.add_stream(StreamConfig(name="MY_STREAM", subjects=["test.*"], storage=StorageType.MEMORY))
subject="test.1"formin ['1', '2', '3']:
awaitjs.publish(subject=subject, payload=m.encode())
asyncdefcb(msg):
logging.info(f'Got {msg}')
sub=awaitjs.subscribe(
subject,
cb=cb,
ordered_consumer=True,
idle_heartbeat=0.5
)
info=awaitsub.consumer_info()
logging.info(f"SUB: {info.name}")
awaitjs.delete_consumer("MY_STREAM", info.name)
awaitasyncio.sleep(3) # now the consumer should reset due to missing HBtry:
info=awaitsub.consumer_info()
logging.info(f"SUB: {info.name}") # should print a new nameexceptnats.js.errors.Errorase:
logging.exception(e)
awaitjs.delete_stream("MY_STREAM")
awaitnc.close()
asyncdefmain():
consumer_task=asyncio.create_task(consume())
try:
awaitasyncio.wait([consumer_task], return_when=asyncio.FIRST_COMPLETED)
exceptExceptionase:
logging.exception(e)
asyncio.run(main())
produces the following output:
C:\playground\nats.py\.venv\Scripts\python.exe C:\playground\nats.py\examples\bar.py
INFO:root:Got Msg(_client=<nats client v2.4.0>, subject='test.1', reply='$JS.ACK.MY_STREAM.KnjEVTb4.1.1.1.1697887152782136300.2', data=b'1', headers=None, _metadata=None, _ackd=False, _sid=2)
INFO:root:Got Msg(_client=<nats client v2.4.0>, subject='test.1', reply='$JS.ACK.MY_STREAM.KnjEVTb4.1.2.2.1697887152782136300.1', data=b'2', headers=None, _metadata=None, _ackd=False, _sid=2)
INFO:root:Got Msg(_client=<nats client v2.4.0>, subject='test.1', reply='$JS.ACK.MY_STREAM.KnjEVTb4.1.3.3.1697887152782675900.0', data=b'3', headers=None, _metadata=None, _ackd=False, _sid=2)
INFO:root:SUB: KnjEVTb4
ERROR:root:nats: NotFoundError: code=404 err_code=10014 description='consumer not found'
Traceback (most recent call last):
File "C:\playground\nats.py\examples\bar.py", line 38, in consume
info = await sub.consumer_info()
^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\playground\nats.py\nats\js\client.py", line 674, in consumer_info
info = await self._js._jsm.consumer_info(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\playground\nats.py\nats\js\manager.py", line 163, in consumer_info
resp = await self._api_request(
^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\playground\nats.py\nats\js\manager.py", line 377, in _api_request
raise APIError.from_error(resp['error'])
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\playground\nats.py\nats\js\errors.py", line 86, in from_error
raise NotFoundError(**err)
nats.js.errors.NotFoundError: nats: NotFoundError: code=404 err_code=10014 description='consumer not found'
Process finished with exit code 0
Expected behavior
The ordered consumer detects missing HB and re-creates the push consumer in NATS; second call to sub.consumer_info() returns updated info. No errors are raised.
Server and client version
NATS Server 2.10.3
nats.py 2.4.0
Host environment
No response
Steps to reproduce
No response
The text was updated successfully, but these errors were encountered:
Observed behavior
Consider the following code:
produces the following output:
Expected behavior
The ordered consumer detects missing HB and re-creates the push consumer in NATS; second call to
sub.consumer_info()
returns updated info. No errors are raised.Server and client version
NATS Server 2.10.3
nats.py 2.4.0
Host environment
No response
Steps to reproduce
No response
The text was updated successfully, but these errors were encountered: