#620889952

Committed 23 Apr 2024 08:23PM UTC coverage: 45.051%. First build

Build # #620889952

Build Type

Pull #26

travis-ci

Pull Request Pull Request #26: Server support for protocols

Run Details

110 of 405 new or added lines in 21 files covered. (27.16%)

7733 of 17165 relevant lines covered (45.05%)

0.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.09

/src/netius/base/protocol.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

# Hive Netius System
# Copyright (c) 2008-2024 Hive Solutions Lda.
#
# This file is part of Hive Netius System.
#
# Hive Netius System is free software: you can redistribute it and/or modify
# it under the terms of the Apache License as published by the Apache
# Foundation, either version 2.0 of the License, or (at your option) any
# later version.
#
# Hive Netius System 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
# Apache License for more details.
#
# You should have received a copy of the Apache License along with
# Hive Netius System. If not, see <http://www.apache.org/licenses/>.

__author__ = "João Magalhães <joamag@hive.pt>"
""" The author(s) of the module """

__copyright__ = "Copyright (c) 2008-2024 Hive Solutions Lda."
""" The copyright for the module """

__license__ = "Apache License, Version 2.0"
""" The license for the module """

from . import legacy
from . import request
from . import observer


class Protocol(observer.Observable):
    """
    Abstract class from which concrete implementation of
    protocol logic should be inherited.

    The logic of a protocol should implement both a reaction
    to the arrival of information (receive) and the sending
    of processed data (send).
    """

    def __init__(self, owner=None):
        observer.Observable.__init__(self)
        self.owner = owner
        self._transport = None
        self._loop = None
        self._writing = True
        self._open = False
        self._closed = False
        self._closing = False
        self._delayed = []
        self._callbacks = []

    def open(self):
        # in case the protocol is already open, ignores the current
        # call as it's considered a double opening
        if self.is_open():
            return

        # calls the concrete implementation of the open operation
        # allowing an extra level of indirection
        self.open_c()

        self.trigger("open", self)

    def close(self):
        # in case the protocol is already closed, ignores the current
        # call considering it a double closing operation
        if self.is_closed() or self.is_closing():
            return

        # calls the concrete implementation of the close operation
        # allowing an extra level of indirection
        self.close_c()

        self.trigger("close", self)

    def finish(self):
        # in case the current protocol is already (completely) closed
        # or is not in the state of closing, nothing should be done
        if self.is_closed():
            return
        if not self.is_closing():
            return

        # calls the concrete implementation of the finish operation
        # allowing an extra level of indirection
        self.finish_c()

        self.trigger("finish", self)

        # runs the "final" destroy operation that is going to run
        # the most structural elements of this object
        self.destroy()

    def open_c(self):
        # unmarks the current protocol from closed (and closing)
        # meaning that it will be opened one more time and
        # so it must not be considered as closed
        self._open = True
        self._closed = False
        self._closing = False

    def close_c(self):
        # marks the current protocol as closing, meaning that although
        # the close operation is not yet finished it's starting
        self._closing = True

        # runs the close transport call that triggers the process
        # of closing the underlying transport method, notice that
        # this operation is only considered to be safely completed
        # on the next tick of the event loop
        self._close_transport()

        # delays the execution of the finish (cleanup) operation
        # so that all the pending operations from the close transport
        # call can be executed in the meantime, ensuring a proper,
        # secure and clean execution of the finish method
        self.delay(self.finish)

    def finish_c(self):
        del self._delayed[:]
        del self._callbacks[:]

        self._transport = None
        self._loop = None
        self._writing = True
        self._open = False
        self._closed = True
        self._closing = False

    def info_dict(self, full=False):
        if not self._transport:
            return dict()
        info = self._transport.info_dict(full=full)
        return info

    def connection_made(self, transport):
        self._transport = transport

        # ensure that the protocol is open, please notice
        # that most of the time the protocol is already open
        self.open()

    def connection_lost(self, exception):
        self.close()

    def transport(self):
        return self._transport

    def loop(self):
        return self._loop

    def loop_set(self, loop):
        self._loop = loop

        self.trigger("loop_set", self)

    def loop_unset(self):
        self._loop = None

        self.trigger("loop_unset", self)

    def pause_writing(self):
        self._writing = False

    def resume_writing(self):
        self._writing = True
        self._flush_callbacks()
        self._flush_send()

    def delay(self, callable, timeout=None):
        # in case there's no event loop defined for the protocol
        # it's not possible to delay this execution so the
        # callable is called immediately
        if not self._loop:
            return callable()

        # verifies if the assigned loop contains the non-standard
        # delay method and if that's the case calls it instead of
        # the base asyncio API ones (compatibility)
        if hasattr(self._loop, "delay"):
            immediately = timeout == None
            return self._loop.delay(callable, timeout=timeout, immediately=immediately)

        # calls the proper call method taking into account if a timeout
        # value exists or not (soon against later)
        if timeout:
            return self._loop.call_later(timeout, callable)
        else:
            return self._loop.call_soon(callable)

    def debug(self, object):
        if not self._loop:
            return
        if not hasattr(self._loop, "debug"):
            return
        self._loop.debug(object)

    def info(self, object):
        if not self._loop:
            return
        if not hasattr(self._loop, "info"):
            return
        self._loop.info(object)

    def warning(self, object):
        if not self._loop:
            return
        if not hasattr(self._loop, "warning"):
            return
        self._loop.warning(object)

    def error(self, object):
        if not self._loop:
            return
        if not hasattr(self._loop, "error"):
            return
        self._loop.error(object)

    def critical(self, object):
        if not self._loop:
            return
        if not hasattr(self._loop, "critical"):
            return
        self._loop.critical(object)

    def is_pending(self):
        return not self._open and not self._closed and not self._closing

    def is_open(self):
        return self._open

    def is_closed(self):
        return self._closed

    def is_closing(self):
        return self._closing

    def is_closed_or_closing(self):
        return self._closed or self._closing

    def is_devel(self):
        if not self._loop:
            return False
        if not hasattr(self._loop, "is_devel"):
            return False
        return self._loop.is_devel()

    def _close_transport(self, force=False):
        if not self._transport:
            return
        self._transport.abort()

    def _delay_send(self, data, address=None, callback=None):
        item = (data, address, callback)
        self._delayed.append(item)
        return len(data)

    def _flush_callbacks(self):
        while self._callbacks:
            callback = self._callbacks.pop(0)
            self.delay(lambda: callback(self._transport))

    def _flush_send(self):
        while True:
            if not self._delayed:
                break
            if not self._writing:
                break
            data, address, callback = self._delayed.pop(0)
            if address:
                self.send(data, address, callback=callback)  # pylint: disable=E1101
            else:
                self.send(data, callback=callback)  # pylint: disable=E1101


class DatagramProtocol(Protocol):

    def __init__(self):
        Protocol.__init__(self)
        self.requests = []
        self.requests_m = {}

    def datagram_received(self, data, address):
        self.on_data(address, data)

    def error_received(self, exception):
        pass

    def on_data(self, address, data):
        self.trigger("data", self, data)

    def send(self, data, address, delay=True, force=False, callback=None):
        return self.send_to(data, address, delay=delay, force=force, callback=callback)

    def send_to(self, data, address, delay=True, force=False, callback=None):
        # ensures that the provided data value is a bytes sequence
        # so that its format is compliant with what's expected by
        # the underlying transport send to operation
        data = legacy.bytes(data)

        # in case the current transport buffers do not allow writing
        # (paused mode) the writing of the data is delayed until the
        # writing is again enabled (resume writing)
        if not self._writing:
            return self._delay_send(data, address=address, callback=callback)

        # pushes the write data down to the transport layer immediately
        # as writing is still allowed for the current protocol
        self._transport.sendto(data, address)

        # in case there's a callback associated with the send
        # tries to see if the data has been completely flushed
        # (writing still enabled) and if so schedules the callback
        # to be called on the next tick, otherwise adds it to the
        # callbacks to be called upon the next write resume operation
        if callback:
            if self._writing:
                self.delay(lambda: callback(self._transport))
            else:
                self._callbacks.append(callback)

        # returns the size (in bytes) of the data that has just been
        # explicitly sent through the associated transport
        return len(data)

    def add_request(self, request):
        # adds the current request object to the list of requests
        # that are pending a valid response, a garbage collector
        # system should be able to erase this request from the
        # pending list in case a timeout value has passed
        self.requests.append(request)
        self.requests_m[request.id] = request

    def remove_request(self, request):
        self.requests.remove(request)
        del self.requests_m[request.id]

    def get_request(self, id):
        is_response = isinstance(id, request.Response)
        if is_response:
            id = id.get_id()
        return self.requests_m.get(id, None)


class StreamProtocol(Protocol):

    def data_received(self, data):
        self.on_data(data)

    def eof_received(self):
        pass

    def on_data(self, data):
        self.trigger("data", self, data)

    def send(self, data, delay=True, force=False, callback=None):
        # ensures that the provided data value is a bytes sequence
        # so that its format is compliant with what's expected by
        # the underlying transport write operation
        data = legacy.bytes(data)

        # in case the current transport buffers do not allow writing
        # (paused mode) the writing of the data is delayed until the
        # writing is again enabled (resume writing)
        if not self._writing:
            return self._delay_send(data, callback=callback)

        # pushes the write data down to the transport layer immediately
        # as writing is still allowed for the current protocol
        self._transport.write(data)

        # in case there's a callback associated with the send
        # tries to see if the data has been completely flushed
        # (writing still enabled), and if so, schedules the callback
        # to be called on the next tick otherwise adds it to the
        # callbacks to be called upon the next write resume operation
        if callback:
            if self._writing:
                self.delay(lambda: callback(self._transport))
            else:
                self._callbacks.append(callback)

        # returns the size (in bytes) of the data that has just been
        # explicitly sent through the associated transport
        return len(data)

1	#!/usr/bin/python
2	# -- coding: utf-8 --
3
4	# Hive Netius System
5	# Copyright (c) 2008-2024 Hive Solutions Lda.
6	#
7	# This file is part of Hive Netius System.
8	#
9	# Hive Netius System is free software: you can redistribute it and/or modify
10	# it under the terms of the Apache License as published by the Apache
11	# Foundation, either version 2.0 of the License, or (at your option) any
12	# later version.
13	#
14	# Hive Netius System is distributed in the hope that it will be useful,
15	# but WITHOUT ANY WARRANTY; without even the implied warranty of
16	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17	# Apache License for more details.
18	#
19	# You should have received a copy of the Apache License along with
20	# Hive Netius System. If not, see <http://www.apache.org/licenses/>.
21
22	__author__ = "João Magalhães <joamag@hive.pt>"	1✔
23	""" The author(s) of the module """
24
25	__copyright__ = "Copyright (c) 2008-2024 Hive Solutions Lda."	1✔
26	""" The copyright for the module """
27
28	__license__ = "Apache License, Version 2.0"	1✔
29	""" The license for the module """
30
31	from . import legacy	1✔
32	from . import request
33	from . import observer
34		1✔
35
36	class Protocol(observer.Observable):
37	"""	1✔
38	Abstract class from which concrete implementation of
39	protocol logic should be inherited.
40		1✔
41	The logic of a protocol should implement both a reaction	1✔
42	to the arrival of information (receive) and the sending	1✔
43	of processed data (send).
44	"""	1✔
45
46	def __init__(self, owner=None):
47	observer.Observable.__init__(self)
48	self.owner = owner
49	self._transport = None
50	self._loop = None
51	self._writing = True
52	self._open = False
53	self._closed = False
54	self._closing = False	1✔
55	self._delayed = []	1✔
56	self._callbacks = []	1✔
57		1✔
58	def open(self):	1✔
59	# in case the protocol is already open, ignores the current	1✔
60	# call as it's considered a double opening	1✔
61	if self.is_open():	1✔
62	return	1✔
63		1✔
64	# calls the concrete implementation of the open operation	1✔
65	# allowing an extra level of indirection
66	self.open_c()	1✔
67
68	self.trigger("open", self)
69		1✔
70	def close(self):
71	# in case the protocol is already closed, ignores the current
72	# call considering it a double closing operation
73	if self.is_closed() or self.is_closing():	1✔
74	return
75		1✔
76	# calls the concrete implementation of the close operation
77	# allowing an extra level of indirection	1✔
78	self.close_c()
79
80	self.trigger("close", self)	1✔
81
82	def finish(self):
83	# in case the current protocol is already (completely) closed
84	# or is not in the state of closing, nothing should be done	1✔
85	if self.is_closed():
86	return	1✔
87	if not self.is_closing():
88	return	1✔
89
90	# calls the concrete implementation of the finish operation
91	# allowing an extra level of indirection	1✔
92	self.finish_c()	1✔
93
94	self.trigger("finish", self)
95
96	# runs the "final" destroy operation that is going to run	1✔
97	# the most structural elements of this object
98	self.destroy()	1✔
99
100	def open_c(self):
101	# unmarks the current protocol from closed (and closing)
102	# meaning that it will be opened one more time and	1✔
103	# so it must not be considered as closed
104	self._open = True	1✔
105	self._closed = False
106	self._closing = False
107
108	def close_c(self):	1✔
109	# marks the current protocol as closing, meaning that although	1✔
110	# the close operation is not yet finished it's starting	1✔
111	self._closing = True
112		1✔
113	# runs the close transport call that triggers the process
114	# of closing the underlying transport method, notice that
115	# this operation is only considered to be safely completed	1✔
116	# on the next tick of the event loop
117	self._close_transport()
118
119	# delays the execution of the finish (cleanup) operation
120	# so that all the pending operations from the close transport
121	# call can be executed in the meantime, ensuring a proper,	1✔
122	# secure and clean execution of the finish method
123	self.delay(self.finish)
124
125	def finish_c(self):
126	del self._delayed[:]
127	del self._callbacks[:]	1✔
128
129	self._transport = None	1✔
130	self._loop = None	1✔
131	self._writing = True	1✔
132	self._open = False
133	self._closed = True	1✔
134	self._closing = False	1✔
135		1✔
136	def info_dict(self, full=False):	1✔
137	if not self._transport:	1✔
138	return dict()	1✔
139	info = self._transport.info_dict(full=full)
140	return info	1✔
141		×
142	def connection_made(self, transport):	×
143	self._transport = transport	×
144
145	# ensure that the protocol is open, please notice	1✔
146	# that most of the time the protocol is already open	1✔
147	self.open()
148
149	def connection_lost(self, exception):
150	self.close()	1✔
151
152	def transport(self):	1✔
153	return self._transport	1✔
154
155	def loop(self):	1✔
156	return self._loop	1✔
157
158	def loop_set(self, loop):	1✔
159	self._loop = loop	×
160
161	self.trigger("loop_set", self)	1✔
162		1✔
163	def loop_unset(self):
164	self._loop = None	1✔
165
166	self.trigger("loop_unset", self)	1✔
167		×
168	def pause_writing(self):
169	self._writing = False	×
170
171	def resume_writing(self):	1✔
172	self._writing = True	×
173	self._flush_callbacks()
174	self._flush_send()	1✔
175		×
176	def delay(self, callable, timeout=None):	×
177	# in case there's no event loop defined for the protocol	×
178	# it's not possible to delay this execution so the
179	# callable is called immediately	1✔
180	if not self._loop:
181	return callable()
182
183	# verifies if the assigned loop contains the non-standard	1✔
184	# delay method and if that's the case calls it instead of
185	# the base asyncio API ones (compatibility)
186	if hasattr(self._loop, "delay"):
187	immediately = timeout == None
188	return self._loop.delay(callable, timeout=timeout, immediately=immediately)	1✔
189		1✔
190	# calls the proper call method taking into account if a timeout	1✔
191	# value exists or not (soon against later)
192	if timeout:
193	return self._loop.call_later(timeout, callable)
194	else:
195	return self._loop.call_soon(callable)
196
197	def debug(self, object):
198	if not self._loop:	×
199	return	×
200	if not hasattr(self._loop, "debug"):
201	return	1✔
202	self._loop.debug(object)	1✔
203		1✔
204	def info(self, object):	1✔
205	if not self._loop:
206	return	1✔
207	if not hasattr(self._loop, "info"):	×
208	return	×
209	self._loop.info(object)	×
210
211	def warning(self, object):	1✔
212	if not self._loop:	×
213	return	×
214	if not hasattr(self._loop, "warning"):	×
215	return
216	self._loop.warning(object)	1✔
217		×
218	def error(self, object):	×
219	if not self._loop:	×
220	return
221	if not hasattr(self._loop, "error"):	1✔
222	return	×
223	self._loop.error(object)	×
224		×
225	def critical(self, object):
226	if not self._loop:	1✔
227	return	×
228	if not hasattr(self._loop, "critical"):
229	return	1✔
230	self._loop.critical(object)	1✔
231
232	def is_pending(self):	1✔
233	return not self._open and not self._closed and not self._closing	1✔
234
235	def is_open(self):	1✔
236	return self._open	1✔
237
238	def is_closed(self):	1✔
239	return self._closed	×
240
241	def is_closing(self):	1✔
242	return self._closing	1✔
243		1✔
244	def is_closed_or_closing(self):	1✔
245	return self._closed or self._closing
246		1✔
247	def is_devel(self):	1✔
248	if not self._loop:	1✔
249	return False
250	if not hasattr(self._loop, "is_devel"):	1✔
251	return False	×
252	return self._loop.is_devel()	×
253		×
254	def _close_transport(self, force=False):
255	if not self._transport:	1✔
256	return	×
257	self._transport.abort()	×
258		×
259	def _delay_send(self, data, address=None, callback=None):
260	item = (data, address, callback)	1✔
261	self._delayed.append(item)	×
262	return len(data)	×
263		×
264	def _flush_callbacks(self):	×
NEW 265	while self._callbacks:	×
NEW 266	callback = self._callbacks.pop(0)	×
267	self.delay(lambda: callback(self._transport))
268		1✔
269	def _flush_send(self):
270	while True:	1✔
271	if not self._delayed:	1✔
272	break	1✔
273	if not self._writing:	1✔
274	break
275	data, address, callback = self._delayed.pop(0)	1✔
276	if address:	1✔
277	self.send(data, address, callback=callback) # pylint: disable=E1101
278	else:	1✔
279	self.send(data, callback=callback) # pylint: disable=E1101	×
280
281		1✔
282	class DatagramProtocol(Protocol):	1✔
283
284	def __init__(self):	1✔
285	Protocol.__init__(self)
286	self.requests = []
287	self.requests_m = {}
288
289	def datagram_received(self, data, address):
290	self.on_data(address, data)
291
292	def error_received(self, exception):	1✔
293	pass
294
295	def on_data(self, address, data):
296	self.trigger("data", self, data)
297
298	def send(self, data, address, delay=True, force=False, callback=None):
299	return self.send_to(data, address, delay=delay, force=force, callback=callback)
300		1✔
301	def send_to(self, data, address, delay=True, force=False, callback=None):
302	# ensures that the provided data value is a bytes sequence
303	# so that its format is compliant with what's expected by
304	# the underlying transport send to operation
305	data = legacy.bytes(data)
306
307	# in case the current transport buffers do not allow writing
308	# (paused mode) the writing of the data is delayed until the
309	# writing is again enabled (resume writing)
310	if not self._writing:
311	return self._delay_send(data, address=address, callback=callback)	1✔
312
313	# pushes the write data down to the transport layer immediately
314	# as writing is still allowed for the current protocol
315	self._transport.sendto(data, address)
316		1✔
317	# in case there's a callback associated with the send	×
318	# tries to see if the data has been completely flushed
319	# (writing still enabled) and if so schedules the callback
320	# to be called on the next tick, otherwise adds it to the
321	# callbacks to be called upon the next write resume operation
322	if callback:
323	if self._writing:
324	self.delay(lambda: callback(self._transport))
325	else:	1✔
326	self._callbacks.append(callback)
327
328	# returns the size (in bytes) of the data that has just been
329	# explicitly sent through the associated transport
330	return len(data)
331
332	def add_request(self, request):	1✔
333	# adds the current request object to the list of requests	×
334	# that are pending a valid response, a garbage collector	×
335	# system should be able to erase this request from the
336	# pending list in case a timeout value has passed
337	self.requests.append(request)
338	self.requests_m[request.id] = request	1✔
339
340	def remove_request(self, request):	1✔
341	self.requests.remove(request)
342	del self.requests_m[request.id]
343
344	def get_request(self, id):
345	is_response = isinstance(id, request.Response)	1✔
346	if is_response:	1✔
347	id = id.get_id()
348	return self.requests_m.get(id, None)	1✔
349		1✔
350		1✔
351	class StreamProtocol(Protocol):
352		1✔
353	def data_received(self, data):	1✔
354	self.on_data(data)	1✔
355		1✔
356	def eof_received(self):
357	pass	1✔
358
359	def on_data(self, data):	1✔
360	self.trigger("data", self, data)	1✔
361
362	def send(self, data, delay=True, force=False, callback=None):	1✔
363	# ensures that the provided data value is a bytes sequence	1✔
364	# so that its format is compliant with what's expected by
365	# the underlying transport write operation	1✔
366	data = legacy.bytes(data)	1✔
367
368	# in case the current transport buffers do not allow writing	1✔
369	# (paused mode) the writing of the data is delayed until the
370	# writing is again enabled (resume writing)
371	if not self._writing:
372	return self._delay_send(data, callback=callback)	1✔
373
374	# pushes the write data down to the transport layer immediately
375	# as writing is still allowed for the current protocol
376	self._transport.write(data)
377		1✔
378	# in case there's a callback associated with the send	×
379	# tries to see if the data has been completely flushed
380	# (writing still enabled), and if so, schedules the callback
381	# to be called on the next tick otherwise adds it to the
382	# callbacks to be called upon the next write resume operation	1✔
383	if callback:
384	if self._writing:
385	self.delay(lambda: callback(self._transport))
386	else:
387	self._callbacks.append(callback)
388
389	# returns the size (in bytes) of the data that has just been	1✔
390	# explicitly sent through the associated transport	1✔
391	return len(data)	×

hivesolutions / netius / #620889952

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous