Repository navigation
How to handle restoring a subscription. #2035
Answered
by
meyshad
iamthebull
asked this question in
Q&A
|
I'm using I believe my options are:
The example program that follows demonstrates this situation. It can be run against import asyncio
import logging
from asyncua.client.client import Client
from asyncua.common.subscription import DataChangeEvent, StatusChangeEvent, Subscription
logging.basicConfig(level=logging.INFO, filename='log/client-example2.log', filemode='w')
_logger = logging.getLogger("asyncua")
async def subscription_handler(subscription: Subscription):
try:
async with subscription:
async for event in subscription:
match event:
case DataChangeEvent(node=node, value=value):
_logger.info("data change %r %s", node, value)
case StatusChangeEvent(notification=notif):
_logger.info("status change %s", notif.Status)
except Exception as e:
_logger.exception(f"error in subscription handler - {e}")
async def main() -> None:
"""
Run against examples/server-example.py
"""
async with asyncio.TaskGroup() as tg:
async with Client(
url="opc.tcp://localhost:4840/freeopcua/server/",
auto_reconnect=True,
reconnect_max_delay=2.0) as client:
idx = await client.get_namespace_index(uri="http://examples.freeopcua.github.io")
var = await client.nodes.objects.get_child(f"{idx}:MyObject/{idx}:MyVariable")
sub = await client.create_subscription(500)
sub_task = tg.create_task(subscription_handler(sub), name='subscription handler')
await sub.subscribe_data_change(var)
await asyncio.sleep(1)
await client.disconnect()
sub_task.cancel()
await client.connect(auto_reconnect=True, reconnect_max_delay=2.0)
await sub.restore()
sub_task = tg.create_task(subscription_handler(sub), name='subscription handler') ### this fails
for i in range(3):
_logger.info(f"tick {i}")
await asyncio.sleep(.5)
sub_task.cancel()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
passOutput: |
Answered by
meyshad
Sep 2, 2026
Replies: 1 comment 1 reply
|
The context manager is the part that deletes the subscription.
I would keep ownership of the subscription outside the consumer task: async def consume(sub):
async for event in sub:
match event:
case DataChangeEvent(node=node, value=value):
_logger.info("data change %r %s", node, value)
case StatusChangeEvent(notification=notif):
_logger.info("status change %s", notif.Status)
sub = await client.create_subscription(500)
await sub.subscribe_data_change(var)
task = asyncio.create_task(consume(sub))
# planned reconnect
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
await client.disconnect()
await client.connect(auto_reconnect=True, reconnect_max_delay=2.0)
await sub.restore()
task = asyncio.create_task(consume(sub))
# final shutdown only
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
await sub.delete()You do not need to copy the monitored items yourself. For an unplanned connection loss, you can also leave the consumer running and let the client's auto-reconnect supervisor restore every live, non-deleted subscription. The main point is not to leave an |
1 reply
|
This is perfect. Much appreciated. |
Answer selected by
iamthebull
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
The context manager is the part that deletes the subscription.
Subscription.__aexit__()always callsdelete(). In this example, cancellingsub_taskunwindsasync with subscription:. Because the transport is already down,delete()logs "local cleanup only", but itsfinallyblock still marks the object as deleted and closes the iterator.restore()cannot bring that object back after_deletedhas been set.I would keep ownership of the subscription outside the consumer task: