Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/publisher.py: 64%

1from fastapi_websocket_pubsub import PubSubClient, Topic 

2from opal_common.confi.confi import load_conf_if_none 

3from opal_common.topics.publisher import ( 

4 ClientSideTopicPublisher, 

5 PeriodicPublisher, 

6 ServerSideTopicPublisher, 

7 TopicPublisher, 

8) 

9from opal_common.utils import get_authorization_header 

10from opal_server.config import opal_server_config 

11 

12 

13def setup_publisher_task( 

14 server_uri: str = None, 

15 server_token: str = None, 

16) -> TopicPublisher: 

17 server_uri = load_conf_if_none( 

18 server_uri, 

19 opal_server_config.OPAL_WS_LOCAL_URL, 

20 ) 

21 server_token = load_conf_if_none( 

22 server_token, 

23 opal_server_config.OPAL_WS_TOKEN, 

24 ) 

25 return ClientSideTopicPublisher( 

26 client=PubSubClient( 

27 additional_headers=[get_authorization_header(server_token)] 

28 ), 

29 server_uri=server_uri, 

30 ) 

31 

32 

33def setup_broadcaster_keepalive_task( 

34 publisher: ServerSideTopicPublisher, 

35 time_interval: int, 

36 topic: Topic = "__broadcast_session_keepalive__", 

37) -> PeriodicPublisher: 

38 """A periodic publisher with the intent to trigger messages on the 

39 broadcast channel, so that the session to the backbone won't become idle 

40 and close on the backbone end.""" 

41 return PeriodicPublisher( 

42 publisher, time_interval, topic, task_name="broadcaster keepalive task" 

43 )