Tell me please: how from service RPyC server refer to the Service specific
client.
*CORE*
class CoreService(rpyc.Service):
"""
Сервис услуг Ядра системы (набор функций, которые можно вызвать
удаленно)
"""
__modules = {}
def __del__(self):
pass
def on_connect(self):
pass
def on_disconnect(self):
for key, value in self.__modules.items():
if id(value) == id(self._conn):
del self.__modules[key]
break
def exposed_register_module(self, module_name):
if module_name in self.__modules:
return False
self.__modules[module_name] = self._conn
return True
def exposed_module_is_connected(self, module_name):
if module_name in self.__modules:
return True
return False
def exposed_execute_query_module(self, module_name, attribute_name, args
):
print '=========================='
print 'MODULE QUERY FROM MODULE:'
print module_name
print attribute_name
print args
print self._conn
print '=========================='
# TTTTTTTTTTHHHHHHHHHIIIIIIIISSSSSSSSSSSSSS
if module_name in self.__modules:
self.__modules[module_name].root
# return None
class Core():
"""
Ядро системы Perseus
"""
__server = None
def __init__(self):
pass
def __del__(self):
self.stop_core()
def start_core(self, address='127.0.0.1', port=33000):
"""
Запуск ядра
:param address: IP адресс модуля
:return: Результат запуска системы
:rtype: bool
"""
self.__server = ThreadedServer(hostname=address, port=port, service=
CoreService)
core_server_thread = threading.Thread(name='core_server_thread',
target=self.__server.start)
core_server_thread.setDaemon(True)
core_server_thread.start()
return True
def stop_core(self):
"""
Функция останавливает Ядро
:return:
"""
self.__server.close()
MODULE
class Module():
"""
"""
__core_connection = None
def __init__(self):
pass
def __del__(self):
self.close()
def connect_to_core(self, address='127.0.0.1', port=33000,
module_service=rpyc.VoidService):
"""
:param address: Адрес, на котором запущено Ядро
:param port: Порт, на котором запущено Ядро
:return: Результат подключения
:rtype: bool
"""
self.__core_connection = rpyc.connect(host=address, port=port,
service=module_service)
if not self.__register_module_in_core(module_service.get_module_name
()):
self.__core_connection.close()
return False
return True
def __register_module_in_core(self, module_name):
"""
Удаленно регистрирует текущий модуль в Ядре
:return:
"""
return self.__core_connection.root.register_module(module_name)
def close(self):
if self.__core_connection is not None:
self.__core_connection.close()
self.__core_connection = None
def get_module(self, module_name):
if self.__core_connection is None:
return None
if self.__core_connection.root.module_is_connected(module_name) is
not None:
return ModuleProxy(core_connector=self.__core_connection,
proxy_module_name=module_name)
return None
class ModuleProxy():
__core_connector = None
__proxy_module_name = None
def __init__(self, core_connector, proxy_module_name):
self.__core_connector = core_connector
self.__proxy_module_name = proxy_module_name
def __getattr__(self, name, *args):
print name
print args
if self.__core_connector is None:
return None
return self.__core_connector.root.execute_query_module(self.
__proxy_module_name, name, args)
class ModuleService(rpyc.Service):
"""
Сервис услуг Модуля системы (набор функций, которые можно вызвать
удаленно)
"""
@staticmethod
def get_module_name():
return 'AbstractModuleService'
TEST
def test_module_interaction(self):
test_core = core.Core()
self.assertTrue(test_core.start_core(port=33003))
test_module1 = module.Module()
self.assertTrue(test_module1.connect_to_core(port=33003,
module_service=TestModuleService1))
test_module2 = module.Module()
self.assertTrue(test_module2.connect_to_core(port=33003,
module_service=TestModuleService2))
remote_module_2 = test_module1.get_module(TestModuleService2.
get_module_name())
self.assertIsNotNone(remote_module_2)
self.assertEqual(remote_module_2.test(), TestModuleService2.
exposed_test())
When you run the test I get in into a loop which is interrupted by a
combination of keys and get the following output:
test
()
==========================
MODULE QUERY FROM MODULE:
TestModule2
test
()
<rpyc.core.protocol.Connection 'conn2' object at 0x7f64e0992490>
==========================
^CTraceback (most recent call last):
File "run_tests.py", line 10, in <module>
unittest.TextTestRunner().run(unittest.TestLoader().
loadTestsFromTestCase(test_interaction.TestInteraction))
File "/usr/lib/python2.7/unittest/runner.py", line 151, in run
test(result)
File "/usr/lib/python2.7/unittest/suite.py", line 70, in __call__
return self.run(*args, **kwds)
File "/usr/lib/python2.7/unittest/suite.py", line 108, in run
test(result)
File "/usr/lib/python2.7/unittest/case.py", line 395, in __call__
return self.run(*args, **kwds)
File "/usr/lib/python2.7/unittest/case.py", line 331, in run
testMethod()
File "/home/kpv/perseus/control-lib/tests/test_interaction.py", line 90,
in test_module_interaction
self.assertEqual(remote_module_2.test(), TestModuleService2.exposed_test
())
File "/home/kpv/perseus/control-lib/perseus_control_lib/module.py", line
67, in __getattr__
return self.__core_connector.root.execute_query_module(self.
__proxy_module_name, name, args)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/netref.py", line
196, in __call__
return syncreq(_self, consts.HANDLE_CALL, args, kwargs)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/netref.py", line 71
, in syncreq
return conn.sync_request(handler, oid, *args)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/protocol.py", line
438, in sync_request
self.serve(0.1)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/protocol.py", line
387, in serve
data = self._recv(timeout, wait_for_lock = True)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/protocol.py", line
344, in _recv
if self._channel.poll(timeout):
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/channel.py", line
43, in poll
return self.stream.poll(timeout)
File "/usr/local/lib/python2.7/dist-packages/rpyc/core/stream.py", line 41
, in poll
rl, _, _ = select([self], [], [], timeout)
KeyboardInterrupt
--
---
You received this message because you are subscribed to the Google Groups
"rpyc" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
For more options, visit https://groups.google.com/d/optout.
