Merge pull request #181 from camunda-community-hub/development

v2.3.2

docs/conf.py

@@ -26,7 +26,7 @@
 author = 'Jonatan Martens'
 
 # The full version, including alpha/beta/rc tags
-release = '2.3.1'
+release = '2.3.2'
 
 # -- General configuration ---------------------------------------------------
 
@@ -59,6 +59,6 @@
 # so a file named "default.css" will overwrite the builtin "default.css".
 html_static_path = ['_static']
 
-version = "2.3.1"
+version = "2.3.2"
 
 master_doc = 'index'

pyzeebe/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "2.3.1"
+__version__ = "2.3.2"
 
 from pyzeebe import exceptions
 from pyzeebe.client.client import ZeebeClient

pyzeebe/grpc_internals/channel_options.py

@@ -0,0 +1,15 @@
+GRPC_CHANNEL_OPTIONS = {
+    # https://docs.camunda.io/docs/product-manuals/zeebe/deployment-guide/operations/setting-up-a-cluster/#keep-alive-intervals
+    # "By default, the official Zeebe clients (Java and Go) send keep alive pings every 45 seconds."
+    "grpc.keepalive_time_ms": 45_000
+}
+
+
+def get_channel_options():
+    """
+    Channel arguments are expected as a tuple of tuples:
+    https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
+    """
+    return tuple(
+        (k, v) for k, v in GRPC_CHANNEL_OPTIONS.items()
+    )

pyzeebe/grpc_internals/zeebe_adapter_base.py

@@ -6,6 +6,7 @@
 
 from pyzeebe.credentials.base_credentials import BaseCredentials
 from pyzeebe.exceptions import ZeebeBackPressure, ZeebeGatewayUnavailable, ZeebeInternalError
+from pyzeebe.grpc_internals.channel_options import get_channel_options
 
 logger = logging.getLogger(__name__)
 
@@ -41,11 +42,11 @@ def _get_connection_uri(hostname: str = None, port: int = None, credentials: Bas
     def _create_channel(connection_uri: str, credentials: BaseCredentials = None,
                         secure_connection: bool = False) -> grpc.Channel:
         if credentials:
-            return grpc.secure_channel(connection_uri, credentials.grpc_credentials)
+            return grpc.secure_channel(connection_uri, credentials.grpc_credentials, options=get_channel_options())
         elif secure_connection:
-            return grpc.secure_channel(connection_uri, grpc.ssl_channel_credentials())
+            return grpc.secure_channel(connection_uri, grpc.ssl_channel_credentials(), options=get_channel_options())
         else:
-            return grpc.insecure_channel(connection_uri)
+            return grpc.insecure_channel(connection_uri, options=get_channel_options())
 
     def _check_connectivity(self, value: grpc.ChannelConnectivity) -> None:
         logger.debug(f"Grpc channel connectivity changed to: {value}")

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="pyzeebe",
-    version="2.3.1",
+    version="2.3.2",
     author="Jonatan Martens",
     author_email="[email protected]",
     description="Zeebe client api",

tests/unit/grpc_internals/channel_options_test.py

@@ -0,0 +1,70 @@
+from copy import deepcopy
+
+import pytest
+from unittest.mock import patch, Mock
+
+from pyzeebe.grpc_internals.zeebe_adapter_base import ZeebeAdapterBase
+
+import pyzeebe.grpc_internals.channel_options
+from pyzeebe.grpc_internals.channel_options import get_channel_options
+
+
+@pytest.fixture
+def revert_monkeypatch_after_test():
+    """
+    This sort of exists in pytest already (docs.pytest.org/en/stable/monkeypatch.html),
+    however that means a bit of "magic" happens, this is a bit clearer and tests the users
+    approach to this.
+    """
+    options_before = deepcopy(pyzeebe.grpc_internals.channel_options.GRPC_CHANNEL_OPTIONS)
+    yield
+    pyzeebe.grpc_internals.channel_options.GRPC_CHANNEL_OPTIONS = options_before
+
+
+def test_get_channel_options_returns_tuple_of_tuple_with_options():
+    assert get_channel_options() == (
+        ("grpc.keepalive_time_ms", 45000),
+    )
+
+
+@pytest.mark.parametrize("grpc_method,call_kwargs",
+                         [
+                             ("grpc.secure_channel", {"secure_connection": True}),
+                             ("grpc.insecure_channel", {"secure_connection": False}),
+                             ("grpc.secure_channel", {"credentials": Mock()}),
+                         ])
+def test_create_channel_called_with_options(grpc_method, call_kwargs, zeebe_adapter):
+    with patch(grpc_method) as channel_mock:
+        ZeebeAdapterBase(**call_kwargs)
+        expected_options = (('grpc.keepalive_time_ms', 45000),)
+        # `call_args.kwargs` as it's not available in python <=3.7
+        # https://docs.python.org/3/library/unittest.mock.html#unittest.mock.Mock.call_args
+        # 0 is args, 1 is kwargs
+        assert channel_mock.call_args[1]["options"] == expected_options
+
+
+@pytest.mark.usefixtures("revert_monkeypatch_after_test")
+def test_monkeypatching_with_options_override():
+    pyzeebe.grpc_internals.channel_options.GRPC_CHANNEL_OPTIONS["grpc.keepalive_time_ms"] = 4000
+    assert get_channel_options() == (
+        ("grpc.keepalive_time_ms", 4000),
+    )
+
+
+@pytest.mark.usefixtures("revert_monkeypatch_after_test")
+def test_monkeypatching_with_options_added():
+    pyzeebe.grpc_internals.channel_options.GRPC_CHANNEL_OPTIONS.update({
+        "grpc.keepalive_timeout_ms": 120000,
+        "grpc.http2.min_time_between_pings_ms": 60000,
+    })
+    assert get_channel_options() == (
+        ("grpc.keepalive_time_ms", 45000),
+        ("grpc.keepalive_timeout_ms", 120000),
+        ("grpc.http2.min_time_between_pings_ms", 60000)
+    )
+
+
+@pytest.mark.usefixtures("revert_monkeypatch_after_test")
+def test_monkeypatching_with_options_removed():
+    pyzeebe.grpc_internals.channel_options.GRPC_CHANNEL_OPTIONS = {}
+    assert get_channel_options() == ()