Subscription object returned by Server or Client objects. The object represent a subscription to an opc-ua server. This is a high level class, especially subscribe_data_change and subscribe_events methods. If more control is necessary look at code and/or use create_monitored_ite
| 72 | |
| 73 | |
| 74 | class Subscription(object): |
| 75 | """ |
| 76 | Subscription object returned by Server or Client objects. |
| 77 | The object represent a subscription to an opc-ua server. |
| 78 | This is a high level class, especially subscribe_data_change |
| 79 | and subscribe_events methods. If more control is necessary look at |
| 80 | code and/or use create_monitored_items method. |
| 81 | """ |
| 82 | |
| 83 | def __init__(self, server, params, handler): |
| 84 | self.logger = logging.getLogger(__name__) |
| 85 | self.server = server |
| 86 | self._client_handle = 200 |
| 87 | self._handler = handler |
| 88 | self.parameters = params # move to data class |
| 89 | self._monitoreditems_map = {} |
| 90 | self._lock = Lock() |
| 91 | self.subscription_id = None |
| 92 | self.has_unknown_handlers = False |
| 93 | response = self.server.create_subscription( |
| 94 | params, self.publish_callback, ready_callback=self.ready_callback) |
| 95 | # Set it here to keep the old behavof as well, but this may not run if |
| 96 | # the above times out |
| 97 | self.subscription_id = response.SubscriptionId |
| 98 | |
| 99 | #Send a publish request so the server has one in its queue |
| 100 | # Servers should alsways be able to handle at least on extra publish request per subscriptions |
| 101 | self.server.publish() |
| 102 | |
| 103 | def delete(self): |
| 104 | """ |
| 105 | Delete subscription on server. This is automatically done by Client and Server classes on exit |
| 106 | """ |
| 107 | results = self.server.delete_subscriptions([self.subscription_id]) |
| 108 | results[0].check() |
| 109 | |
| 110 | def is_ready(self): |
| 111 | return bool(self.subscription_id) |
| 112 | |
| 113 | def ready_callback(self, response): |
| 114 | self.subscription_id = self.subscription_id or response.Parameters.SubscriptionId |
| 115 | self.server.publish() |
| 116 | |
| 117 | def publish_callback(self, publishresult): |
| 118 | self.logger.info("Publish callback called with result: %s", publishresult) |
| 119 | if not self.is_ready(): |
| 120 | self.logger.warning( |
| 121 | "Result received but subscription not ready %s", publishresult) |
| 122 | return |
| 123 | |
| 124 | if publishresult.NotificationMessage.NotificationData is not None: |
| 125 | for notif in publishresult.NotificationMessage.NotificationData: |
| 126 | if isinstance(notif, ua.DataChangeNotification): |
| 127 | self._call_datachange(notif) |
| 128 | elif isinstance(notif, ua.EventNotificationList): |
| 129 | self._call_event(notif) |
| 130 | elif isinstance(notif, ua.StatusChangeNotification): |
| 131 | self._call_status(notif) |
no outgoing calls
no test coverage detected