Return ServiceFault for bad subscription ID and return ack results in PublishResult

asyncua/common/subscription.py

@@ -17,6 +17,7 @@
 from asyncua import ua
 from asyncua.client.ua_session import UaSession
 from asyncua.common.ua_utils import copy_dataclass_attr
+from asyncua.common.utils import ServiceError
 from asyncua.ua.uaerrors import BadMessageNotAvailable
 
 if TYPE_CHECKING:
@@ -596,7 +597,10 @@ async def recreate(self) -> None:
         params.SubscriptionId = self.subscription_id
         params.ItemsToCreate = mirs
         params.TimestampsToReturn = ua.TimestampsToReturn.Both
-        results = await self.server.create_monitored_items(params)
+        try:
+            results = await self.server.create_monitored_items(params)
+        except ServiceError as e:
+            raise ua.UaStatusCodeError(e.code)
         for idx, result in enumerate(results):
             mi = params.ItemsToCreate[idx]
             assert mi.RequestedParameters.ClientHandle is not None
@@ -850,7 +854,10 @@ async def create_monitored_items(
             data.monitoring_mode = mi.MonitoringMode
             data.sampling_interval = mi.RequestedParameters.SamplingInterval
             self._monitored_items[mi.RequestedParameters.ClientHandle] = data
-        results = await self.server.create_monitored_items(params)
+        try:
+            results = await self.server.create_monitored_items(params)
+        except ServiceError as e:
+            raise ua.UaStatusCodeError(e.code)
         mids = []
         # process result, add server_handle, or remove it if failed
         for idx, result in enumerate(results):
@@ -876,7 +883,10 @@ async def unsubscribe(self, handle: int | Iterable[int]) -> None:
         params = ua.DeleteMonitoredItemsParameters()
         params.SubscriptionId = self.subscription_id
         params.MonitoredItemIds = list(handles)
-        results = await self.server.delete_monitored_items(params)
+        try:
+            results = await self.server.delete_monitored_items(params)
+        except ServiceError as e:
+            raise ua.UaStatusCodeError(e.code)
         results[0].check()
         handle_map = {v.server_handle: k for k, v in self._monitored_items.items()}
         for handle in handles:
@@ -917,7 +927,10 @@ async def modify_monitored_item(
         params = ua.ModifyMonitoredItemsParameters()
         params.SubscriptionId = self.subscription_id
         params.ItemsToModify.append(modif_item)
-        results = await self.server.modify_monitored_items(params)
+        try:
+            results = await self.server.modify_monitored_items(params)
+        except ServiceError as e:
+            raise ua.UaStatusCodeError(e.code)
         item_to_change.mfilter = results[0].FilterResult
         return results
 

asyncua/server/internal_session.py

@@ -325,7 +325,7 @@ async def delete_monitored_items(
 
         return subscription_result
 
-    def publish(self, acks: Iterable[ua.SubscriptionAcknowledgement] | None = None) -> int:
+    def publish(self, acks: Iterable[ua.SubscriptionAcknowledgement] | None = None) -> tuple[int, list[ua.StatusCode]]:
         return self.subscription_service.publish(acks or [])
 
     def modify_subscription(

asyncua/server/internal_subscription.py

@@ -2,20 +2,25 @@
 server side implementation of a subscription object
 """
 
+from __future__ import annotations
+
 import asyncio
 import logging
 import time
 from collections.abc import Awaitable, Callable, Iterable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from asyncua import ua
 
 from .address_space import AddressSpace
 from .monitored_item_service import MonitoredItemService
 
-PublishResultCallback = Callable[..., Awaitable[None]]
-PublishRequestCallback = Callable[[int], Any]
-DeleteCallback = Callable[[], None]
+if TYPE_CHECKING:
+    from asyncua.server.uaprocessor import PublishRequestData
+
+    PublishResultCallback = Callable[..., Awaitable[None]]
+    PublishRequestCallback = Callable[[int], PublishRequestData | None]
+    DeleteCallback = Callable[[], Any]
 
 
 class InternalSubscription:
@@ -132,7 +137,7 @@ def has_published_results(self) -> bool:
         self._keep_alive_count += 1
         return False
 
-    async def publish_results(self, requestdata: Any = None) -> bool:
+    async def publish_results(self, requestdata: PublishRequestData | None = None) -> bool:
         """
         Publish all enqueued data changes, events and status changes though the callback.
         This method gets first called without publish request from subscription loop.

asyncua/server/subscription_service.py

@@ -111,49 +111,40 @@ async def delete_subscriptions(self, ids: list[int]) -> list[ua.StatusCode]:
                 self.logger.warning("Exception while stopping subscription", exc_info=stop_result)
         return res
 
-    def publish(self, acks: Iterable[ua.SubscriptionAcknowledgement]) -> int:
+    def publish(self, acks: Iterable[ua.SubscriptionAcknowledgement]) -> tuple[int, list[ua.StatusCode]]:
         self.logger.info("publish request with acks %s", acks)
         if not self.subscriptions:
             raise utils.ServiceError(ua.StatusCodes.BadNoSubscription)
+        results: list[ua.StatusCode] = []
         for ack in acks:
             sub = self.subscriptions.get(ack.SubscriptionId)
-            if sub is not None:
+            if sub is None:
+                results.append(ua.StatusCode(ua.StatusCodes.BadSubscriptionIdInvalid))
+            else:
                 sub.publish([ack.SequenceNumber])
-        return len(self.subscriptions)
+                results.append(ua.StatusCode())
+        return len(self.subscriptions), results
 
     async def create_monitored_items(
         self, params: ua.CreateMonitoredItemsParameters
     ) -> list[ua.MonitoredItemCreateResult]:
         self.logger.info("create monitored items")
         if params.SubscriptionId not in self.subscriptions:
-            res: list[ua.MonitoredItemCreateResult] = []
-            for _ in params.ItemsToCreate:
-                response = ua.MonitoredItemCreateResult()
-                response.StatusCode = ua.StatusCode(ua.StatusCodes.BadSubscriptionIdInvalid)
-                res.append(response)
-            return res
+            raise utils.ServiceError(ua.StatusCodes.BadSubscriptionIdInvalid)
         return await self.subscriptions[params.SubscriptionId].monitored_item_srv.create_monitored_items(params)
 
     def modify_monitored_items(
         self, params: ua.ModifyMonitoredItemsParameters
     ) -> list[ua.MonitoredItemModifyResult]:
         self.logger.info("modify monitored items")
         if params.SubscriptionId not in self.subscriptions:
-            res: list[ua.MonitoredItemModifyResult] = []
-            for _ in params.ItemsToModify:
-                result = ua.MonitoredItemModifyResult()
-                result.StatusCode = ua.StatusCode(ua.StatusCodes.BadSubscriptionIdInvalid)
-                res.append(result)
-            return res
+            raise utils.ServiceError(ua.StatusCodes.BadSubscriptionIdInvalid)
         return self.subscriptions[params.SubscriptionId].monitored_item_srv.modify_monitored_items(params)
 
     def delete_monitored_items(self, params: ua.DeleteMonitoredItemsParameters) -> list[ua.StatusCode]:
         self.logger.info("delete monitored items")
         if params.SubscriptionId not in self.subscriptions:
-            res: list[ua.StatusCode] = []
-            for _ in params.MonitoredItemIds:
-                res.append(ua.StatusCode(ua.StatusCodes.BadSubscriptionIdInvalid))
-            return res
+            raise utils.ServiceError(ua.StatusCodes.BadSubscriptionIdInvalid)
         return self.subscriptions[params.SubscriptionId].monitored_item_srv.delete_monitored_items(
             params.MonitoredItemIds
         )

asyncua/server/uaprocessor.py

@@ -17,9 +17,10 @@
 
 
 class PublishRequestData:
-    def __init__(self, requesthdr=None, seqhdr=None):
+    def __init__(self, requesthdr=None, seqhdr=None, results: list[ua.StatusCode] | None=None):
         self.requesthdr = requesthdr
         self.seqhdr = seqhdr
+        self.results = results
         self.timestamp = time.monotonic()
 
     def has_timed_out(self, now: float) -> bool:
@@ -128,6 +129,7 @@ async def forward_publish_response(self, result: ua.PublishResult, requestdata:
         # _logger.info("forward publish response %s", result)
         response = ua.PublishResponse()
         response.Parameters = result
+        result.Results = requestdata.results
         self.send_response(requestdata.requesthdr.RequestHandle, requestdata.seqhdr, response)
 
     async def process(self, header, body):
@@ -496,8 +498,8 @@ async def _process_message(self, typeid, requesthdr, seqhdr, body):
                 if not self.session:
                     return False
                 params = struct_from_binary(ua.PublishParameters, body)
-                subscriptions = self.session.publish(params.SubscriptionAcknowledgements)
-                data = PublishRequestData(requesthdr=requesthdr, seqhdr=seqhdr)
+                subscriptions, results = self.session.publish(params.SubscriptionAcknowledgements)
+                data = PublishRequestData(requesthdr=requesthdr, seqhdr=seqhdr, results=results)
                 # If there is an enqueued publish results callback, try to call it immediately
                 while self._publish_results_subs:
                     subscription_id = next(iter(self._publish_results_subs))