diff options
Diffstat (limited to 'deluge')
-rw-r--r-- | deluge/core/rpcserver.py | 62 | ||||
-rw-r--r-- | deluge/transfer.py | 170 | ||||
-rw-r--r-- | deluge/ui/client.py | 191 |
3 files changed, 273 insertions, 150 deletions
diff --git a/deluge/core/rpcserver.py b/deluge/core/rpcserver.py index b9cfba378..3ecbd39e8 100644 --- a/deluge/core/rpcserver.py +++ b/deluge/core/rpcserver.py @@ -60,6 +60,8 @@ from deluge.core.authmanager import (AUTH_LEVEL_NONE, AUTH_LEVEL_DEFAULT, from deluge.error import (DelugeError, NotAuthorizedError, WrappedException, _ClientSideRecreateError, IncompatibleClient) +from deluge.transfer import DelugeTransferProtocol + RPC_RESPONSE = 1 RPC_ERROR = 2 RPC_EVENT = 3 @@ -134,54 +136,34 @@ class ServerContextFactory(object): ctx.use_privatekey_file(os.path.join(ssl_dir, "daemon.pkey")) return ctx -class DelugeRPCProtocol(Protocol): - __buffer = None +class DelugeRPCProtocol(DelugeTransferProtocol): - def dataReceived(self, data): + def message_received(self, request): """ - This method is called whenever data is received from a client. The + This method is called whenever a message is received from a client. The only message that a client sends to the server is a RPC Request message. If the RPC Request message is valid, then the method is called in :meth:`dispatch`. - - :param data: the data from the client. It should be a zlib compressed - rencoded string. - :type data: str + + :param request: the request from the client. + :type data: tuple """ - if self.__buffer: - # We have some data from the last dataReceived() so lets prepend it - data = self.__buffer + data - self.__buffer = None - - while data: - dobj = zlib.decompressobj() - try: - request = rencode.loads(dobj.decompress(data)) - except Exception, e: - #log.debug("Received possible invalid message (%r): %s", data, e) - # This could be cut-off data, so we'll save this in the buffer - # and try to prepend it on the next dataReceived() - self.__buffer = data - return - else: - data = dobj.unused_data - - if type(request) is not tuple: - log.debug("Received invalid message: type is not tuple") - return + if type(request) is not tuple: + log.debug("Received invalid message: type is not tuple") + return - if len(request) < 1: - log.debug("Received invalid message: there are no items") - return + if len(request) < 1: + log.debug("Received invalid message: there are no items") + return - for call in request: - if len(call) != 4: - log.debug("Received invalid rpc request: number of items " - "in request is %s", len(call)) - continue - #log.debug("RPCRequest: %s", format_request(call)) - reactor.callLater(0, self.dispatch, *call) + for call in request: + if len(call) != 4: + log.debug("Received invalid rpc request: number of items " + "in request is %s", len(call)) + continue + #log.debug("RPCRequest: %s", format_request(call)) + reactor.callLater(0, self.dispatch, *call) def sendData(self, data): """ @@ -192,7 +174,7 @@ class DelugeRPCProtocol(Protocol): :type data: object """ - self.transport.write(zlib.compress(rencode.dumps(data))) + self.transfer_message(data) def connectionMade(self): """ diff --git a/deluge/transfer.py b/deluge/transfer.py new file mode 100644 index 000000000..c89f89e4f --- /dev/null +++ b/deluge/transfer.py @@ -0,0 +1,170 @@ +# -*- coding: utf-8 -*- +# +# transfer.py +# +# Copyright (C) 2012 Bro <bro.development@gmail.com> +# +# Deluge is free software. +# +# You may 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. +# +# deluge 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 deluge. If not, write to: +# The Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor +# Boston, MA 02110-1301, USA. +# +# In addition, as a special exception, the copyright holders give +# permission to link the code of portions of this program with the OpenSSL +# library. +# You must obey the GNU General Public License in all respects for all of +# the code used other than OpenSSL. If you modify file(s) with this +# exception, you may extend this exception to your version of the file(s), +# but you are not obligated to do so. If you do not wish to do so, delete +# this exception statement from your version. If you delete this exception +# statement from all source files in the program, then also delete it here. +# +# + +try: + import rencode +except ImportError: + import deluge.rencode as rencode + +import zlib +import struct + +from twisted.internet.protocol import Protocol + +from deluge.log import LOG as log + +MESSAGE_HEADER_SIZE = 5 + +class DelugeTransferProtocol(Protocol): + """ + Data messages are transfered using very a simple protocol. + Data messages are transfered with a header containing + the length of the data to be transfered (payload). + + """ + def __init__(self): + self._buffer = "" + self._message_length = 0 + self._bytes_received = 0 + self._bytes_sent = 0 + + def transfer_message(self, data): + """ + Transfer the data. + + The data will be serialized and compressed before being sent. + First a header is sent - containing the length of the compressed payload + to come as a signed integer. After the header, the payload is transfered. + + :param data: data to be transfered in a data structure serializable by rencode. + + """ + compressed = zlib.compress(rencode.dumps(data)) + size_data = len(compressed) + # Store length as a signed integer (using 4 bytes). "!" denotes network byte order. + payload_len = struct.pack("!i", size_data) + header = "D" + payload_len + self._bytes_sent += len(header) + len(compressed) + self.transport.write(header) + self.transport.write(compressed) + + def dataReceived(self, data): + """ + This method is called whenever data is received. + + :param data: a message as transfered by transfer_message, or a part of such + a messsage. + + Global variables: + _buffer - contains the data received + _message_length - the length of the payload of the current message. + + """ + self._buffer += data + self._bytes_received += len(data) + + while len(self._buffer) >= MESSAGE_HEADER_SIZE: + if self._message_length == 0: + self._handle_new_message() + # We have a complete packet + if len(self._buffer) >= self._message_length: + self._handle_complete_message(self._buffer[:self._message_length]) + # Remove message data from buffer + self._buffer = self._buffer[self._message_length:] + self._message_length = 0 + else: + break + + def _handle_new_message(self): + """ + Handle the start of a new message. This method is called only when the + beginning of the buffer contains data from a new message (i.e. the header). + + """ + try: + # Read the first bytes of the message (MESSAGE_HEADER_SIZE bytes) + header = self._buffer[:MESSAGE_HEADER_SIZE] + payload_len = header[1:MESSAGE_HEADER_SIZE] + if header[0] != 'D': + raise Exception("Invalid header format. First byte is %d" % ord(header[0])) + # Extract the length stored as a signed integer (using 4 bytes) + self._message_length = struct.unpack("!i", payload_len)[0] + if self._message_length < 0: + raise Exception("Message length is negative: %d" % self._message_length) + # Remove the header from the buffer + self._buffer = self._buffer[MESSAGE_HEADER_SIZE:] + except Exception, e: + log.warn("Error occured when parsing message header: %s." % str(e)) + log.warn("This version of Deluge cannot communicate with the sender of this data.") + self._message_length = 0 + self._buffer = "" + + def _handle_complete_message(self, data): + """ + Handles a complete message as it is transfered on the network. + + :param data: a zlib compressed string encoded with rencode. + + """ + try: + self.message_received(rencode.loads(zlib.decompress(data))) + except Exception, e: + log.warn("Failed to decompress (%d bytes) and load serialized data "\ + "with rencode: %s" % (len(data), str(e))) + + def get_bytes_recv(self): + """ + Returns the number of bytes received. + + :returns: the number of bytes received + :rtype: int + + """ + return self._bytes_received + + def get_bytes_sent(self): + """ + Returns the number of bytes sent. + + :returns: the number of bytes sent + :rtype: int + + """ + return self._bytes_sent + + def message_received(self, message): + """Override this method to receive the complete message""" + pass diff --git a/deluge/ui/client.py b/deluge/ui/client.py index 8be7aaed5..479b7367d 100644 --- a/deluge/ui/client.py +++ b/deluge/ui/client.py @@ -35,18 +35,13 @@ # import logging -from twisted.internet.protocol import Protocol, ClientFactory +from twisted.internet.protocol import ClientFactory from twisted.internet import reactor, ssl, defer -try: - import rencode -except ImportError: - import deluge.rencode as rencode - -import zlib import deluge.common from deluge import error from deluge.event import known_events +from deluge.transfer import DelugeTransferProtocol if deluge.common.windows_check(): import win32api @@ -104,11 +99,11 @@ class DelugeRPCRequest(object): return (self.request_id, self.method, self.args, self.kwargs) -class DelugeRPCProtocol(Protocol): + +class DelugeRPCProtocol(DelugeTransferProtocol): def connectionMade(self): self.__rpc_requests = {} - self.__buffer = None # Set the protocol in the daemon so it can send data self.factory.daemon.protocol = self # Get the address of the daemon that we've connected to @@ -116,102 +111,80 @@ class DelugeRPCProtocol(Protocol): self.factory.daemon.host = peer.host self.factory.daemon.port = peer.port self.factory.daemon.connected = True - log.info("Connected to daemon at %s:%s..", peer.host, peer.port) self.factory.daemon.connect_deferred.callback((peer.host, peer.port)) - def dataReceived(self, data): + def message_received(self, request): """ - This method is called whenever we receive data from the daemon. + This method is called whenever we receive a message from the daemon. + + :param request: a tuple that should be either a RPCResponse, RCPError or RPCSignal - :param data: a zlib compressed and rencoded string that should be either - a RPCResponse, RCPError or RPCSignal """ - if self.__buffer: - # We have some data from the last dataReceived() so lets prepend it - data = self.__buffer + data - self.__buffer = None + if type(request) is not tuple: + log.debug("Received invalid message: type is not tuple") + return + if len(request) < 3: + log.debug("Received invalid message: number of items in " + "response is %s", len(request)) + return + + message_type = request[0] - while data: - # Increase the byte counter - self.factory.bytes_recv += len(data) + if message_type == RPC_EVENT: + event = request[1] + #log.debug("Received RPCEvent: %s", event) + # A RPCEvent was received from the daemon so run any handlers + # associated with it. + if event in self.factory.event_handlers: + for handler in self.factory.event_handlers[event]: + reactor.callLater(0, handler, *request[2]) + return + + request_id = request[1] + + # We get the Deferred object for this request_id to either run the + # callbacks or the errbacks dependent on the response from the daemon. + d = self.factory.daemon.pop_deferred(request_id) + + if message_type == RPC_RESPONSE: + # Run the callbacks registered with this Deferred object + d.callback(request[2]) + elif message_type == RPC_ERROR: + # Recreate exception and errback'it + exception_cls = getattr(error, request[2]) + exception = exception_cls(*request[3], **request[4]) + + # Ideally we would chain the deferreds instead of instance + # checking just to log them. But, that would mean that any + # errback on the fist deferred should returns it's failure + # so it could pass back to the 2nd deferred on the chain. But, + # that does not always happen. + # So, just do some instance checking and just log rpc error at + # diferent levels. + r = self.__rpc_requests[request_id] + msg = "RPCError Message Received!" + msg += "\n" + "-" * 80 + msg += "\n" + "RPCRequest: " + r.__repr__() + msg += "\n" + "-" * 80 + if isinstance(exception, error.WrappedException): + msg += "\n" + exception.type + "\n" + exception.message + ": " + msg += exception.traceback + else: + msg += "\n" + request[5] + "\n" + request[2] + ": " + msg += str(exception) + msg += "\n" + "-" * 80 - dobj = zlib.decompressobj() - try: - request = rencode.loads(dobj.decompress(data)) - except Exception, e: - #log.debug("Received possible invalid message (%r): %s", data, e) - # This could be cut-off data, so we'll save this in the buffer - # and try to prepend it on the next dataReceived() - self.__buffer = data - return + if not isinstance(exception, error._ClientSideRecreateError): + # Let's log these as errors + log.error(msg) else: - data = dobj.unused_data - - if type(request) is not tuple: - log.debug("Received invalid message: type is not tuple") - return - if len(request) < 3: - log.debug("Received invalid message: number of items in " - "response is %s", len(3)) - return - - message_type = request[0] - - if message_type == RPC_EVENT: - event = request[1] - #log.debug("Received RPCEvent: %s", event) - # A RPCEvent was received from the daemon so run any handlers - # associated with it. - if event in self.factory.event_handlers: - for handler in self.factory.event_handlers[event]: - reactor.callLater(0, handler, *request[2]) - continue - - request_id = request[1] - - # We get the Deferred object for this request_id to either run the - # callbacks or the errbacks dependent on the response from the daemon. - d = self.factory.daemon.pop_deferred(request_id) - - if message_type == RPC_RESPONSE: - # Run the callbacks registered with this Deferred object - d.callback(request[2]) - elif message_type == RPC_ERROR: - # Recreate exception and errback'it - exception_cls = getattr(error, request[2]) - exception = exception_cls(*request[3], **request[4]) - - # Ideally we would chain the deferreds instead of instance - # checking just to log them. But, that would mean that any - # errback on the fist deferred should returns it's failure - # so it could pass back to the 2nd deferred on the chain. But, - # that does not always happen. - # So, just do some instance checking and just log rpc error at - # diferent levels. - r = self.__rpc_requests[request_id] - msg = "RPCError Message Received!" - msg += "\n" + "-" * 80 - msg += "\n" + "RPCRequest: " + r.__repr__() - msg += "\n" + "-" * 80 - if isinstance(exception, error.WrappedException): - msg += "\n" + exception.type + "\n" + exception.message + ": " - msg += exception.traceback - else: - msg += "\n" + request[5] + "\n" + request[2] + ": " - msg += str(exception) - msg += "\n" + "-" * 80 - - if not isinstance(exception, error._ClientSideRecreateError): - # Let's log these as errors - log.error(msg) - else: - # The rest just get's logged in debug level, just to log - # what's happening - log.debug(msg) - - d.errback(exception) - del self.__rpc_requests[request_id] + # The rest just get's logged in debug level, just to log + # what's happening + log.debug(msg) + + d.errback(exception) + del self.__rpc_requests[request_id] def send_request(self, request): """ @@ -220,15 +193,16 @@ class DelugeRPCProtocol(Protocol): :param request: RPCRequest """ - # Store the DelugeRPCRequest object just in case a RPCError is sent in - # response to this request. We use the extra information when printing - # out the error for debugging purposes. - self.__rpc_requests[request.request_id] = request - #log.debug("Sending RPCRequest %s: %s", request.request_id, request) - # Send the request in a tuple because multiple requests can be sent at once - data = zlib.compress(rencode.dumps((request.format_message(),))) - self.factory.bytes_sent += len(data) - self.transport.write(data) + try: + # Store the DelugeRPCRequest object just in case a RPCError is sent in + # response to this request. We use the extra information when printing + # out the error for debugging purposes. + self.__rpc_requests[request.request_id] = request + #log.debug("Sending RPCRequest %s: %s", request.request_id, request) + # Send the request in a tuple because multiple requests can be sent at once + self.transfer_message((request.format_message(),)) + except Exception, e: + log.warn("Error occured when sending message:" + str(e)) class DelugeRPCClientFactory(ClientFactory): protocol = DelugeRPCProtocol @@ -237,9 +211,6 @@ class DelugeRPCClientFactory(ClientFactory): self.daemon = daemon self.event_handlers = event_handlers - self.bytes_recv = 0 - self.bytes_sent = 0 - def startedConnecting(self, connector): log.info("Connecting to daemon at \"%s:%s\"...", connector.host, connector.port) @@ -462,10 +433,10 @@ class DaemonSSLProxy(DaemonProxy): self.disconnect_callback = cb def get_bytes_recv(self): - return self.__factory.bytes_recv + return self.protocol.get_bytes_recv() def get_bytes_sent(self): - return self.__factory.bytes_sent + return self.protocol.get_bytes_sent() class DaemonClassicProxy(DaemonProxy): def __init__(self, event_handlers=None): |