Fix to timeout server stream if remote endpoint isn't available. (#10)

* Fix to timeout server stream if remote endpoint isn't available.

Author: rlubke

.github/workflows/validate.yml

@@ -16,8 +16,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10.x", "3.11.x"]
-        poetry-version: ["1.4.2"]
+        python-version: ["3.11.x"]
+        poetry-version: ["1.5.0"]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
     steps:

.pre-commit-config.yaml

@@ -26,7 +26,7 @@ repos:
         - id: black
 
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.2.0
+    rev: v1.3.0
     hooks:
         - id: mypy
 

DEVELOPING.md

@@ -9,8 +9,8 @@ http://oss.oracle.com/licenses/upl.
 
 ### Pre-Requisites
 * Use `bash` shell
-* [pyenv](https://github.com/pyenv/pyenv) version 2.2.4 or later
-* [poetry](https://github.com/python-poetry/poetry) version 1.1.13 or later
+* [pyenv](https://github.com/pyenv/pyenv) version 2.3.x or later
+* [poetry](https://github.com/python-poetry/poetry) version 1.5.x or later
 
 ### Project Structure
 * `bin` - various shell scripts
@@ -19,18 +19,18 @@ http://oss.oracle.com/licenses/upl.
 * `tests` - contains the library test cases in plain Python
 
 ### Setup working environment
-1. Install Python version 3.10.1
+1. Install Python version 3.11.3
 
-    ```pyenv install 3.10.1```
+    ```pyenv install 3.11.3```
 
 2. Checkout the source from GitHub and change directory to the project root dir
 
-3. Set pyenv local version for this project to 3.10.1
+3. Set pyenv local version for this project to  3.11.3
 
-    ```pyenv local 3.10.1```
-4. Set poetry to use python 3.10.1
+    ```pyenv local  3.11.3```
+4. Set poetry to use python  3.11.3
 
-    ```poetry env use 3.10.1```
+    ```poetry env use 3.11.3```
 5. Install all the required dependencies using poetry
 
     ```poetry install```

README.md

@@ -20,7 +20,8 @@ The Coherence Python Client allows Python applications to act as cache clients t
 
 #### Requirements
 * [Coherence CE](https://github.com/oracle/coherence) 22.06.4+ or Coherence 14.1.1.2206.4+ Commercial edition with a configured [gRPCProxy](https://docs.oracle.com/en/middleware/standalone/coherence/14.1.1.2206/develop-remote-clients/using-coherence-grpc-server.html).
-* Python 3.10.1
+* Python 3.11.x
+
 
 #### Starting a Coherence Cluster
 
@@ -56,7 +57,7 @@ import asyncio
 async def run_test():
 
     # create a new Session to the Coherence server
-    session: Session = Session(None)
+    session: Session = await Session.create()
 
     # create a new NamedCache with key of string|int and value of string|int
     cache: NamedCache[str, str|int] = await session.get_cache("test")

examples/aggregators.py

@@ -36,7 +36,7 @@ async def do_run() -> None:
         5: Hobbit(3, "Peregrin", 28, "Side Kick"),
     }
 
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         namedMap: NamedMap[int, Hobbit] = await session.get_map("aggregation-test")
 

examples/basics.py

@@ -14,7 +14,7 @@ async def do_run() -> None:
 
     :return: None
     """
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         namedMap: NamedMap[int, str] = await session.get_map("my-map")
 

examples/events.py

@@ -15,7 +15,7 @@ async def do_run() -> None:
 
     :return: None
     """
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         namedMap: NamedMap[int, str] = await session.get_map("listeners-map")
         await namedMap.put(1, "1")

examples/filters.py

@@ -28,7 +28,7 @@ async def do_run() -> None:
 
     :return: None
     """
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         homes: List[str] = ["Hobbiton", "Buckland", "Frogmorton", "Stock"]
         namedMap: NamedMap[int, Hobbit] = await session.get_map("hobbits")

examples/processors.py

@@ -26,7 +26,7 @@ async def do_run() -> None:
 
     :return: None
     """
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         namedMap: NamedMap[int, Hobbit] = await session.get_map("hobbits")
 

examples/python_object_keys_and_values.py

@@ -29,7 +29,7 @@ async def do_run() -> None:
 
     :return: None
     """
-    session: Session = Session()
+    session: Session = await Session.create()
     try:
         namedMap: NamedMap[AccountKey, Account] = await session.get_map("accounts")
 

pyproject.toml

@@ -21,24 +21,24 @@ classifiers = [
 ]
 
 [tool.poetry.dependencies]
-python = "~3.10"
-protobuf = "~4.21"
-grpcio = "~1.51"
-grpcio-tools = "~1.51"
-jsonpickle = "~2.2"
+python = "~3.11"
+protobuf = "~4.23"
+grpcio = "~1.54"
+grpcio-tools = "~1.54"
+jsonpickle = "~3.0"
 pymitter = "~0.4"
 
 [tool.poetry.dev-dependencies]
-pytest = "~7.2"
-pytest-asyncio = "~0.19"
-pytest-cov = "~4.0"
+pytest = "~7.3"
+pytest-asyncio = "~0.21"
+pytest-cov = "~4.1"
 pytest-unordered = "~0.5"
-pre-commit = "~2.20"
+pre-commit = "~3.3"
 Sphinx = "~4.5"
-sphinx-rtd-theme = "~1.1"
+sphinx-rtd-theme = "~1.2"
 sphinxcontrib-napoleon = "~0.7"
 m2r = "~0.3"
-third-party-license-file-generator = "~2022.3"
+third-party-license-file-generator = "~2023.2"
 
 [tool.pytest.ini_options]
 pythonpath = ["src"]

src/coherence/__init__.py

@@ -6,6 +6,8 @@
 
 __version__ = "1.0b1"
 
+import logging
+
 # expose these symbols in top-level namespace
 from .aggregator import Aggregators as Aggregators
 from .client import MapEntry as MapEntry
@@ -16,3 +18,11 @@
 from .client import TlsOptions as TlsOptions
 from .filter import Filters as Filters
 from .processor import Processors as Processors
+
+# default logging configuration for coherence
+handler: logging.StreamHandler = logging.StreamHandler()  # type: ignore
+handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
+
+COH_LOG = logging.getLogger("coherence")
+COH_LOG.setLevel(logging.INFO)
+COH_LOG.addHandler(handler)

src/coherence/client.py

@@ -6,8 +6,9 @@
 
 import abc
 import asyncio
+import logging
 import os
-from asyncio import Task
+from asyncio import Condition, Task
 from threading import Lock
 from typing import (
     Any,
@@ -45,6 +46,8 @@
 R = TypeVar("R")
 T = TypeVar("T")
 
+COH_LOG = logging.getLogger("coherence")
+
 
 @no_type_check
 def _pre_call_cache(func):
@@ -58,6 +61,9 @@ async def inner_async(self, *args, **kwargs):
         if not self.active:
             raise Exception("Cache [{}] has been {}.".format(self.name, "released" if self.released else "destroyed"))
 
+        # noinspection PyProtectedMember
+        await self._session._wait_for_active()
+
         return await func(self, *args, **kwargs)
 
     if asyncio.iscoroutinefunction(func):
@@ -499,6 +505,7 @@ def __init__(self, cache_name: str, session: Session, serializer: Serializer):
         self._internal_emitter: EventEmitter = EventEmitter()
         self._destroyed: bool = False
         self._released: bool = False
+        self._session: Session = session
         from .event import _MapEventsManager
 
         self._setup_event_handlers()
@@ -743,16 +750,18 @@ def _setup_event_handlers(self) -> None:
         # noinspection PyProtectedMember
         def on_destroyed(name: str) -> None:
             if name == cache_name:
-                this._events_manager._close()
-                this._destroyed = True
-                emitter.emit(MapLifecycleEvent.DESTROYED.value, name)
+                if not this.destroyed:
+                    this._events_manager._close()
+                    this._destroyed = True
+                    emitter.emit(MapLifecycleEvent.DESTROYED.value, name)
 
         # noinspection PyProtectedMember
         def on_released(name: str) -> None:
             if name == cache_name:
-                this._events_manager._close()
-                this._released = True
-                emitter.emit(MapLifecycleEvent.RELEASED.value, name)
+                if not this.released:
+                    this._events_manager._close()
+                    this._released = True
+                    emitter.emit(MapLifecycleEvent.RELEASED.value, name)
 
         def on_truncated(name: str) -> None:
             if name == cache_name:
@@ -922,7 +931,7 @@ def __init__(
           corresponding `ConfigurableCacheFactory` on the server.
         :param request_timeout_seconds: Defines the request timeout, in `seconds`, that will be applied to each
           remote call. If not explicitly set, this defaults to :func:`coherence.client.Options.DEFAULT_REQUEST_TIMEOUT`.
-          See also See also :func:`coherence.client.Options.ENV_REQUEST_TIMEOUT`
+          See also :func:`coherence.client.Options.ENV_REQUEST_TIMEOUT`
         :param ser_format: The serialization format.  Currently, this is always `json`
         :param channel_options: The `gRPC` `ChannelOptions`. See
             https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments and
@@ -941,7 +950,8 @@ def __init__(
             try:
                 time_out = float(timeout)
             except ValueError:
-                print(f"The value of {Options.ENV_REQUEST_TIMEOUT} cannot be converted to a float")
+                COH_LOG.warning("The timeout value of [%s] cannot be converted to a float", Options.ENV_REQUEST_TIMEOUT)
+
             self._request_timeout_seconds = time_out
         else:
             self._request_timeout_seconds = request_timeout_seconds
@@ -1080,6 +1090,8 @@ def __init__(self, session_options: Optional[Options] = None):
         :param session_options: the provided :func:`coherence.client.Options`
         """
         self._closed: bool = False
+        self._active = False
+        self._active_condition: Condition = Condition()
         self._caches: dict[str, NamedCache[Any, Any]] = dict()
         self._lock: Lock = Lock()
         if session_options is not None:
@@ -1121,6 +1133,13 @@ def __init__(self, session_options: Optional[Options] = None):
         watch_task: Task[None] = asyncio.create_task(watch_channel_state(self))
         self._tasks.add(watch_task)
         self._emitter: EventEmitter = EventEmitter()
+        self._channel.get_state(True)  # trigger connect
+
+    @staticmethod
+    async def create(session_options: Optional[Options] = None) -> Session:
+        session: Session = Session(session_options)
+        await session._set_active(False)
+        return session
 
     # noinspection PyTypeHints
     @_pre_call_session
@@ -1236,6 +1255,30 @@ async def get_map(self, name: str, ser_format: str = DEFAULT_FORMAT) -> "NamedMa
                 self._caches.update({name: c})
             return c
 
+    def is_active(self) -> bool:
+        """
+        Returns
+        :return:
+        """
+        return self._active
+
+    async def _set_active(self, active: bool) -> None:
+        self._active = active
+        if self._active:
+            if not self._active_condition.locked():
+                await self._active_condition.acquire()
+            self._active_condition.notify_all()
+            self._active_condition.release()
+        else:
+            await self._active_condition.acquire()
+
+    async def _wait_for_active(self) -> None:
+        if not self.is_active():
+            timeout: float = self._session_options.request_timeout_seconds
+            COH_LOG.debug("Waiting for session to become active; timeout=[%s seconds]", timeout)
+            async with asyncio.timeout(timeout):
+                await self._active_condition.wait()
+
     # noinspection PyUnresolvedReferences
     async def close(self) -> None:
         """
@@ -1248,6 +1291,10 @@ async def close(self) -> None:
                 task.cancel()
             self._tasks.clear()
 
+            caches_copy: dict[str, NamedCache[Any, Any]] = self._caches.copy()
+            for cache in caches_copy.values():
+                await cache.destroy()
+
             await self._channel.close()  # TODO: consider grace period?
 
     def _setup_event_handlers(self, client: NamedCacheClient[K, V]) -> None:
@@ -1342,28 +1389,37 @@ async def watch_channel_state(session: Session) -> None:
     emitter: EventEmitter = session._emitter
     channel: grpc.aio.Channel = session.channel
     first_connect: bool = True
-    connected: bool = False
+    last_state: grpc.ChannelConnectivity = grpc.ChannelConnectivity.IDLE
 
     try:
         while True:
             state: grpc.ChannelConnectivity = channel.get_state(False)
+            COH_LOG.debug("New Channel State: transitioning from [%s] to [%s]", last_state, state)
             match state:
                 case grpc.ChannelConnectivity.SHUTDOWN:
-                    continue  # nothing to do
+                    COH_LOG.info("Session to [%s] terminated", session.options.address)
+                    await session._set_active(False)
                 case grpc.ChannelConnectivity.READY:
-                    if not first_connect and not connected:
+                    if not first_connect and not session.is_active():
+                        COH_LOG.info("Session re-established to [%s]", session.options.address)
+
                         await emitter.emit_async(SessionLifecycleEvent.RECONNECTED.value)
-                        connected = True
-                    elif first_connect and not connected:
+                        await session._set_active(True)
+                    elif first_connect and not session.is_active():
+                        COH_LOG.info("Session established to [%s]", session.options.address)
+
                         first_connect = False
-                        connected = True
                         await emitter.emit_async(SessionLifecycleEvent.CONNECTED.value)
+                        await session._set_active(True)
                 case _:
-                    if connected:
+                    if session.is_active():
+                        COH_LOG.warning("Session to [%s] disconnected; will attempt reconnect", session.options.address)
+
                         await emitter.emit_async(SessionLifecycleEvent.DISCONNECTED.value)
-                        connected = False
+                        await session._set_active(False)
 
-            await channel.wait_for_state_change(state)
+            COH_LOG.debug("Waiting for state change from [%s]", state)
+            await channel.wait_for_state_change(channel.get_state(True))
     except asyncio.CancelledError:
         return