1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
# Copyright (C) 2012- Takafumi Arakaki
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
from .py3compat import SimpleXMLRPCServer
def _get_logger():
"""
Generate a logger with a stream handler.
"""
logger = logging.getLogger('epc')
hndlr = logging.StreamHandler()
hndlr.setLevel(logging.INFO)
hndlr.setFormatter(logging.Formatter(logging.BASIC_FORMAT))
logger.addHandler(hndlr)
return logger
_logger = _get_logger()
class EPCDispatcher:
# This class will be mixed with `SocketServer.TCPServer`,
# which is an old style class.
# see also: SimpleXMLRPCServer.SimpleXMLRPCDispatcher
def __init__(self):
self.funcs = {}
self.instance = None
def register_instance(self, instance, allow_dotted_names=False):
"""
Register an instance to respond to EPC requests.
:type instance: object
:arg instance:
An object with methods to provide to peer. If this
instance has `_get_method` method, EPC method name
resolution can be done by this method.
:type allow_dotted_names: bool
:arg allow_dotted_names:
If it is true, method names containing dots are supported.
They are resolved using `getattr` for each part of the
name as long as it does not start with '_'.
Unlike :meth:`register_function`, only one instance can
be registered.
"""
self.instance = instance
self.allow_dotted_names = allow_dotted_names
def register_function(self, function, name=None):
"""
Register function to be called from EPC client.
:type function: callable
:arg function: Function to publish.
:type name: str
:arg name: Name by which function is published.
This method returns the given `function` as-is, so that you
can use it as a decorator.
"""
if name is None:
name = function.__name__
self.funcs[name] = function
return function
def get_method(self, name):
"""
Get registered method callend `name`.
"""
try:
return self.funcs[name]
except KeyError:
try:
return self.instance._get_method(name)
except AttributeError:
return SimpleXMLRPCServer.resolve_dotted_attribute(
self.instance, name, self.allow_dotted_names)
class EPCCore(EPCDispatcher):
"""
Core methods shared by `EPCServer` and `EPCClient`.
"""
logger = _logger
def __init__(self, debugger, log_traceback):
EPCDispatcher.__init__(self)
self.set_debugger(debugger)
self.log_traceback = log_traceback
def set_debugger(self, debugger):
"""
Set debugger to run when an error occurs in published method.
You can also set debugger by passing `debugger` argument to
the class constructor.
:type debugger: {'pdb', 'ipdb', None}
:arg debugger: type of debugger.
"""
if debugger == 'pdb':
import pdb
self.debugger = pdb
elif debugger == 'ipdb':
import ipdb
self.debugger = ipdb
else:
self.debugger = debugger
|
