X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=ricsdl-package%2Fricsdl%2Fsyncstorage.py;h=62e3a7f0f726bf964d6b6a091c61f63e44e6e94f;hb=refs%2Ftags%2F3.1.0;hp=55063e486f192b49ec66600b06979233e3217baf;hpb=db7753971931247abf7fed67921074e518ec1f6d;p=ric-plt%2Fsdlpy.git diff --git a/ricsdl-package/ricsdl/syncstorage.py b/ricsdl-package/ricsdl/syncstorage.py index 55063e4..62e3a7f 100755 --- a/ricsdl-package/ricsdl/syncstorage.py +++ b/ricsdl-package/ricsdl/syncstorage.py @@ -1,5 +1,5 @@ # Copyright (c) 2019 AT&T Intellectual Property. -# Copyright (c) 2018-2019 Nokia. +# Copyright (c) 2018-2022 Nokia. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -119,6 +119,7 @@ class SyncStorage(SyncStorageAbc): """ def __init__(self, fake_db_backend=None) -> None: super().__init__() + self.__dbbackend = None self.__configuration = _Configuration(fake_db_backend) self.event_separator = self.__configuration.get_event_separator() self.__dbbackend = ricsdl.backend.get_backend_instance(self.__configuration) @@ -141,7 +142,8 @@ class SyncStorage(SyncStorageAbc): return False def close(self): - self.__dbbackend.close() + if self.__dbbackend: + self.__dbbackend.close() @func_arg_checker(SdlTypeError, 1, ns=str, data_map=dict) def set(self, ns: str, data_map: Dict[str, bytes]) -> None: @@ -266,8 +268,7 @@ class SyncStorage(SyncStorageAbc): self.__dbbackend.remove_all_and_publish(ns, channels_and_events) @func_arg_checker(SdlTypeError, 1, ns=str, cb=Callable, channels=(str, builtins.set)) - def subscribe_channel(self, ns: str, - cb: Union[Callable[[str, str], None], Callable[[str, List[str]], None]], + def subscribe_channel(self, ns: str, cb: Callable[[str, List[str]], None], channels: Union[str, Set[str]]) -> None: self._validate_callback(cb) channels = [channels] if isinstance(channels, str) else list(channels) @@ -281,7 +282,7 @@ class SyncStorage(SyncStorageAbc): def start_event_listener(self) -> None: self.__dbbackend.start_event_listener() - def handle_events(self) -> Optional[Union[Tuple[str, str], Tuple[str, List[str]]]]: + def handle_events(self) -> Optional[Tuple[str, List[str]]]: return self.__dbbackend.handle_events() @func_arg_checker(SdlTypeError, 1, ns=str, resource=str, expiration=(int, float))