Crossbar json serializer

golem/node.py

@@ -10,6 +10,7 @@
     Dict,
     List,
     Optional,
+    TYPE_CHECKING,
     TypeVar,
 )
 
@@ -48,6 +49,11 @@
 from golem.tools.uploadcontroller import UploadController
 from golem.tools.remotefs import RemoteFS
 
+if TYPE_CHECKING:
+    # pylint:disable=unused-import
+    from golem.rpc.router import SerializerType
+
+
 F = TypeVar('F', bound=Callable[..., Any])
 logger = logging.getLogger(__name__)
 
@@ -87,7 +93,8 @@ def __init__(self,  # noqa pylint: disable=too-many-arguments
                  use_talkback: bool = False,
                  use_docker_manager: bool = True,
                  geth_address: Optional[str] = None,
-                 password: Optional[str] = None
+                 password: Optional[str] = None,
+                 crossbar_serializer: 'Optional[SerializerType]' = None,
                 ) -> None:
 
         # DO NOT MAKE THIS IMPORT GLOBAL
@@ -154,6 +161,8 @@ def __init__(self,  # noqa pylint: disable=too-many-arguments
             if not self.set_password(password):
                 raise Exception("Password incorrect")
 
+        self._crossbar_serializer = crossbar_serializer
+
     def start(self) -> None:
 
         HardwarePresets.initialize(self._datadir)
@@ -275,6 +284,7 @@ def _start_rpc(self) -> Deferred:
             host=self._config_desc.rpc_address,
             port=self._config_desc.rpc_port,
             datadir=self._datadir,
+            crossbar_serializer=self._crossbar_serializer,
         )
         self._reactor.addSystemEventTrigger("before", "shutdown", rpc.stop)
 

golem/rpc/router.py

@@ -1,10 +1,9 @@
+import enum
 import json
 import logging
 import os
-from collections import namedtuple
 from typing import Iterable, Optional
 
-import enum
 from crossbar.common import checkconfig
 from twisted.internet.defer import inlineCallbacks
 
@@ -16,16 +15,18 @@
 
 logger = logging.getLogger('golem.rpc.crossbar')
 
-CrossbarRouterOptions = namedtuple(
-    'CrossbarRouterOptions',
-    ['cbdir', 'logdir', 'loglevel', 'argv', 'config']
-)
+
+@enum.unique
+class SerializerType(enum.Enum):
+    def _generate_next_value_(name, *_):  # pylint: disable=no-self-argument
+        return name
+
+    json = enum.auto()
+    msgpack = enum.auto()
 
 
 # pylint: disable=too-many-instance-attributes
 class CrossbarRouter(object):
-    serializers = ['msgpack']
-
     @enum.unique
     class CrossbarRoles(enum.Enum):
         admin = enum.auto()
@@ -37,10 +38,9 @@ def __init__(self,
                  host: Optional[str] = CROSSBAR_HOST,
                  port: Optional[int] = CROSSBAR_PORT,
                  realm: str = CROSSBAR_REALM,
-                 crossbar_log_level: str = 'info',
                  ssl: bool = True,
-                 generate_secrets: bool = False) -> None:
-
+                 generate_secrets: bool = False,
+                 crossbar_serializer: Optional[SerializerType] = None) -> None:
         self.working_dir = os.path.join(datadir, CROSSBAR_DIR)
 
         os.makedirs(self.working_dir, exist_ok=True)
@@ -53,27 +53,27 @@ def __init__(self,
 
         self.address = WebSocketAddress(host, port, realm, ssl)
 
-        self.log_level = crossbar_log_level
         self.node = None
         self.pubkey = None
 
-        self.options = self._build_options()
+        if crossbar_serializer is None:
+            crossbar_serializer = SerializerType.msgpack
+
         self.config = self._build_config(self.address,
-                                         self.serializers,
+                                         [crossbar_serializer.name],
                                          self.cert_manager)
 
         logger.debug('xbar init with cfg: %s', json.dumps(self.config))
 
-    def start(self, reactor, options=None):
+    def start(self, reactor):
         # imports reactor
         from crossbar.controller.node import Node, default_native_workers
 
-        options = options or self.options
         if self.address.ssl:
             self.cert_manager.generate_if_needed()
 
-        self.node = Node(options.cbdir, reactor=reactor)
-        self.pubkey = self.node.maybe_generate_key(options.cbdir)
+        self.node = Node(self.working_dir, reactor=reactor)
+        self.pubkey = self.node.maybe_generate_key(self.working_dir)
 
         workers = default_native_workers()
 
@@ -85,15 +85,6 @@ def start(self, reactor, options=None):
     def stop(self):
         yield self.node._controller.shutdown()  # noqa # pylint: disable=protected-access
 
-    def _build_options(self, argv=None, config=None):
-        return CrossbarRouterOptions(
-            cbdir=self.working_dir,
-            logdir=None,
-            loglevel=self.log_level,
-            argv=argv,
-            config=config
-        )
-
     @staticmethod
     def _users_config(cert_manager: cert.CertificateManager):
         # configuration for crsb_users with admin priviliges

golemapp.py

@@ -27,6 +27,7 @@
 from golem.core import variables  # noqa
 from golem.core.common import install_reactor  # noqa
 from golem.core.simpleenv import get_local_datadir  # noqa
+from golem.rpc.router import SerializerType  # noqa
 
 logger = logging.getLogger('golemapp')  # using __name__ gives '__main__' here
 
@@ -104,6 +105,12 @@ def monkey_patched_getLogger(*args, **kwargs):
 @click.option('--enable-talkback', is_flag=True, default=None)
 @click.option('--hyperdrive-port', type=int, help="Hyperdrive public port")
 @click.option('--hyperdrive-rpc-port', type=int, help="Hyperdrive RPC port")
+@click.option('--crossbar-serializer', default=None,
+              type=click.Choice([
+                  SerializerType.msgpack.value,
+                  SerializerType.json.value,
+              ]),
+              help="Crossbar serializer (default: msgpack)")
 # Python flags, needed by crossbar (package only)
 @click.option('-m', nargs=1, default=None)
 @click.option('--node', expose_value=False)
@@ -122,7 +129,7 @@ def start(  # pylint: disable=too-many-arguments, too-many-locals
         monitor, concent, datadir, node_address, rpc_address, peer, mainnet,
         net, geth_address, password, accept_terms, accept_concent_terms,
         accept_all_terms, version, log_level, enable_talkback, m,
-        hyperdrive_port, hyperdrive_rpc_port,
+        hyperdrive_port, hyperdrive_rpc_port, crossbar_serializer
 ):
 
     freeze_support()
@@ -197,6 +204,8 @@ def _start():
             concent_variant=ethereum_config.CONCENT_VARIANT,
             geth_address=geth_address,
             password=password,
+            crossbar_serializer=(SerializerType(crossbar_serializer)
+                                 if crossbar_serializer else None),
         )
 
         if accept_terms:

scripts/node_integration_tests/nodes/json_serializer.py

@@ -0,0 +1,15 @@
+from golemapp import start
+from golem.client import Client
+from golem.rpc import utils as rpc_utils
+
+
+@rpc_utils.expose('test.bignum')
+def _get_bignum(self):
+    return 2**64 + 1337
+
+
+# using setattr silences mypy complaining about "has no attribute"
+setattr(Client, "_get_bignum", _get_bignum)
+
+
+start()

scripts/node_integration_tests/playbooks/golem/json_serializer/playbook.py

@@ -0,0 +1,26 @@
+from functools import partial
+
+from ...base import NodeTestPlaybook
+from ...test_config_base import NodeId
+
+
+class Playbook(NodeTestPlaybook):
+    def step_check_bignum(self):
+        def on_success(result):
+            if result != (2**64 + 1337):
+                self.fail()
+                return
+            print("transferring bigints works correctly")
+            self.next()
+
+        def on_error(error):
+            print(f"Error: {error}")
+            self.fail()
+
+        return self.call(NodeId.requestor, 'test.bignum', on_success=on_success,
+                         on_error=on_error)
+
+    steps = (
+        partial(NodeTestPlaybook.step_get_key, node_id=NodeId.requestor),
+        step_check_bignum,
+    ) + NodeTestPlaybook.steps

scripts/node_integration_tests/playbooks/golem/json_serializer/test_config.py

@@ -0,0 +1,13 @@
+from ...test_config_base import TestConfigBase, NodeId
+
+
+class TestConfig(TestConfigBase):
+    def __init__(self):
+        super().__init__()
+        self.nodes[NodeId.requestor].script = 'json_serializer'
+        # if you remove crossbar-serializer flag below, test should fail with
+        # "WAMP message serialization error: huge unsigned int".
+        for node_config in self.nodes.values():
+            node_config.additional_args = {
+                '--crossbar-serializer': 'json',
+            }

scripts/node_integration_tests/playbooks/test_config_base.py

@@ -20,6 +20,7 @@
 
 class NodeConfig:
     def __init__(self) -> None:
+        self.additional_args: Dict[str, Any] = {}
         self.concent = 'staging'
         # if datadir is None it will be automatically created
         self.datadir: Optional[str] = None
@@ -51,6 +52,7 @@ def make_args(self) -> Dict[str, Any]:
             args['--hyperdrive-port'] = self.hyperdrive_port
         if self.hyperdrive_rpc_port:
             args['--hyperdrive-rpc-port'] = self.hyperdrive_rpc_port
+        args.update(self.additional_args)
 
         return args
 

tests/golem/test_opt_node.py

@@ -138,7 +138,8 @@ def test_geth_address_should_be_passed_to_node(self, mock_node, *_):
                                      ],
                                      use_monitor=None,
                                      use_talkback=None,
-                                     password=None)
+                                     password=None,
+                                     crossbar_serializer=None)
 
     @patch('golem.node.TransactionSystem')
     def test_geth_address_should_be_passed_to_transaction_system(
@@ -213,7 +214,8 @@ def test_mainnet_should_be_passed_to_node(self, mock_node, *_):
                                      concent_variant=concent_disabled,
                                      use_monitor=None,
                                      use_talkback=None,
-                                     password=None)
+                                     password=None,
+                                     crossbar_serializer=None)
 
     @patch('golem.node.Client')
     def test_mainnet_should_be_passed_to_client(self, mock_client, *_):
@@ -262,7 +264,8 @@ def test_net_testnet_should_be_passed_to_node(self, mock_node, *_):
             concent_variant=variables.CONCENT_CHOICES['test'],
             use_monitor=None,
             use_talkback=None,
-            password=None
+            password=None,
+            crossbar_serializer=None,
         )
 
     @patch('golem.node.Node')
@@ -290,7 +293,8 @@ def test_net_mainnet_should_be_passed_to_node(self, mock_node, *_):
                                      concent_variant=concent_disabled,
                                      use_monitor=None,
                                      use_talkback=None,
-                                     password=None)
+                                     password=None,
+                                     crossbar_serializer=None)
 
     @patch('golem.node.Node')
     def test_config_change(self, *_):