Coverage for /home/runner/work/zserio/zserio/compiler/extensions/python/runtime/src/zserio/pubsub.py: 100%

1""" 

2The module provides classes for Zserio Pub/Sub. 

3""" 

4 

5import typing 

6 

7from zserio.exception import PythonRuntimeException 

8 

9 

10class PubsubInterface: 

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

12 

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

14 """ 

15 Publishes given data as a specified topic. 

16 

17 :param topic: Topic definition. 

18 :param data: Data to publish. 

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

20 :raises PubsubException: If the publishing fails. 

21 """ 

22 raise NotImplementedError() 

23 

24 def subscribe( 

25 self, 

26 topic: str, 

27 callback: typing.Callable[[str, bytes], None], 

28 context: typing.Any = None, 

29 ) -> int: 

30 """ 

31 Subscribes a topic. 

32 

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

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

35 supported by Pub/Sub backend. 

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

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

38 :returns: Subscription ID. 

39 :raises PubsubException: If subscribing fails. 

40 """ 

41 raise NotImplementedError() 

42 

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

44 """ 

45 Unsubscribes the subscription with the given ID. 

46 

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

48 :raises PubsubException: If unsubscribing fails. 

49 """ 

50 raise NotImplementedError() 

51 

52 

53class PubsubException(PythonRuntimeException): 

54 """ 

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

56 """