Coverage for /home/jenkins/workspace/NDS/Zserio/NDS_ZSERIO-linux-build/compiler/extensions/python/runtime/src/zserio/pubsub.py: 100%

10 statements  

« prev     ^ index     » next       coverage.py v6.5.0, created at 2023-12-13 15:12 +0000

1""" 

2The module provides classes for Zserio Pub/Sub. 

3""" 

4 

5import typing 

6 

7from zserio.exception import PythonRuntimeException 

8 

9class PubsubInterface: 

10 """ Interface for Pub/Sub client backends. """ 

11 

12 def publish(self, topic: str, data: bytes, context: typing.Any = None) -> None: 

13 """ 

14 Publishes given data as a specified topic. 

15 

16 :param topic: Topic definition. 

17 :param data: Data to publish. 

18 :param context: Context specific for a particular Pub/Sub implementation. 

19 :raises PubsubException: If the publishing fails. 

20 """ 

21 raise NotImplementedError() 

22 

23 def subscribe(self, topic: str, callback: typing.Callable[[str, bytes], None], 

24 context: typing.Any = None) -> int: 

25 """ 

26 Subscribes a topic. 

27 

28 :param topic: Topic definition to subscribe. Note that the definition format depends on the particular 

29 Pub/Sub backend implementation and therefore e.g. wildcards can be used only if they are 

30 supported by Pub/Sub backend. 

31 :param callback: Callback to be called when a message with the specified topic arrives. 

32 :param context: Context specific for a particular Pub/Sub implementation. 

33 :returns: Subscription ID. 

34 :raises PubsubException: If subscribing fails. 

35 """ 

36 raise NotImplementedError() 

37 

38 def unsubscribe(self, subscription_id: int) -> None: 

39 """ 

40 Unsubscribes the subscription with the given ID. 

41 

42 :param id: ID of the subscription to be unsubscribed. 

43 :raises PubsubException: If unsubscribing fails. 

44 """ 

45 raise NotImplementedError() 

46 

47class PubsubException(PythonRuntimeException): 

48 """ 

49 Exception thrown in case of an error in Zserio Pub/Sub. 

50 """