~bzr-pqm/bzr/bzr.dev

#

# 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

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

 

"""Wire-level encoding and decoding of requests and responses for the smart

client and server.

"""

 

import collections

from cStringIO import StringIO

import struct

import sys

import time

 

import bzrlib

from bzrlib.smart import message, request

from bzrlib.trace import log_exception_quietly, mutter

 

 

# Protocol version strings.  These are sent as prefixes of bzr requests and

 

def _encode_tuple(args):

    """Encode the tuple args to a bytestream."""

 

 

class Requester(object):

        for start, length in offsets:

            txt.append('%d,%d' % (start, length))

        return '\n'.join(txt)

 

class SmartServerRequestProtocolOne(SmartProtocolBase):

    """Server-side encoding and decoding logic for smart version 1."""

        self._backing_transport = backing_transport

        self._root_client_path = root_client_path

        self.unused_data = ''

        self._finished = False

        self.in_buffer = ''

 

    def accept_bytes(self, bytes):

        """Take bytes, and advance the internal state machine appropriately.

        :param bytes: must be a byte string

        """

        if not isinstance(bytes, str):

                req_args = _decode_tuple(first_line)

                self.request = request.SmartServerRequestHandler(

                    self._backing_transport, commands=request.request_handlers,

                if self.request.finished_reading:

                    # trivial request

                    self.unused_data = self.in_buffer

 

        if self._has_dispatched:

            if self._finished:

                # machine too.

                self.unused_data += self.in_buffer

                self.in_buffer = ''

 

    def _write_protocol_version(self):

        """Write any prefixes this protocol requires.

        Version one doesn't send protocol versions.

        """

 

 

class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):

    r"""Version two of the server side of the smart protocol.

    This prefixes responses with the value of RESPONSE_VERSION_TWO.

    """

 

 

    def _write_protocol_version(self):

        r"""Write any prefixes this protocol requires.

        Version two sends the value of RESPONSE_VERSION_TWO.

        """

        self._write_func(self.response_marker)

 

    def __init__(self):

        self.finished_reading = False

        self.unused_data = ''

        self.bytes_left = None

        self._number_needed_bytes = None

 

    def accept_bytes(self, bytes):

        """Decode as much of bytes as possible.

 

        data will be appended to self.unused_data.

        """

        # accept_bytes is allowed to change the state

        self._number_needed_bytes = None

        try:

            # Run the function for the current state.

            self.state_accept()

            while current_state != self.state_accept:

                # The current state has changed.  Run the function for the new

        self.chunks = collections.deque()

        self.error = False

        self.error_in_progress = None

    def next_read_size(self):

        # Note: the shortest possible chunk is 2 bytes: '0\n', and the

        # end-of-body marker is 4 bytes: 'END\n'.

            # the rest of this chunk plus an END chunk.

            return self.bytes_left + 4

        elif self.state_accept == self._state_accept_expecting_length:

                # We're expecting a chunk length.  There's at least two bytes

                # left: a digit plus '\n'.

                return 2

        elif self.state_accept == self._state_accept_reading_unused:

            return 1

        elif self.state_accept == self._state_accept_expecting_header:

        else:

            raise AssertionError("Impossible state: %r" % (self.state_accept,))

 

            return None

 

    def _extract_line(self):

        if pos == -1:

            # We haven't read a complete line yet, so request more bytes before

            # we continue.

            raise _NeedMoreBytes(1)

        # Trim the prefix (including '\n' delimiter) from the _in_buffer.

        return line

 

    def _finished(self):

        self.state_accept = self._state_accept_reading_unused

        if self.error:

            error_args = tuple(self.error_in_progress)

            self.state_accept = self._state_accept_reading_chunk

 

    def _state_accept_reading_chunk(self):

        self.bytes_left -= in_buffer_len

        if self.bytes_left <= 0:

            # Finished with chunk

                self.chunks.append(self.chunk_in_progress)

            self.chunk_in_progress = None

            self.state_accept = self._state_accept_expecting_length

    def _state_accept_reading_unused(self):

 

 

class LengthPrefixedBodyDecoder(_StatefulDecoder):

    """Decodes the length-prefixed bulk data."""

    def __init__(self):

        _StatefulDecoder.__init__(self)

        self.state_accept = self._state_accept_expecting_length

        self.state_read = self._state_read_no_data

        self._body = ''

        self._trailer_buffer = ''

    def next_read_size(self):

        if self.bytes_left is not None:

            # Ideally we want to read all the remainder of the body and the

        else:

            # Reading excess data.  Either way, 1 byte at a time is fine.

            return 1

    def read_pending_data(self):

        """Return any pending data that has been decoded."""

        return self.state_read()

 

    def _state_accept_expecting_length(self):

        if pos == -1:

            return

        self.state_accept = self._state_accept_reading_body

        self.state_read = self._state_read_body_buffer

 

    def _state_accept_reading_body(self):

        if self.bytes_left <= 0:

            # Finished with body

            if self.bytes_left != 0:

                self._body = self._body[:self.bytes_left]

            self.bytes_left = None

            self.state_accept = self._state_accept_reading_trailer

    def _state_accept_reading_trailer(self):

        # TODO: what if the trailer does not match "done\n"?  Should this raise

        # a ProtocolViolation exception?

        if self._trailer_buffer.startswith('done\n'):

            self.unused_data = self._trailer_buffer[len('done\n'):]

            self.state_accept = self._state_accept_reading_unused

            self.finished_reading = True

    def _state_accept_reading_unused(self):

 

    def _state_read_no_data(self):

        return ''

            mutter('hpss call:   %s', repr(args)[1:-1])

            if getattr(self._request._medium, 'base', None) is not None:

                mutter('             (to %s)', self._request._medium.base)

        self._write_args(args)

        self._request.finished_writing()

        self._last_verb = args[0]

            if getattr(self._request._medium, '_path', None) is not None:

                mutter('                  (to %s)', self._request._medium._path)

            mutter('              %d bytes', len(body))

            if 'hpssdetail' in debug.debug_flags:

                mutter('hpss body content: %s', body)

        self._write_args(args)

        """Make a remote call with a readv array.

 

        The body is encoded with one line per readv offset pair. The numbers in

        """

        if 'hpss' in debug.debug_flags:

            mutter('hpss call w/readv: %s', repr(args)[1:-1])

            if getattr(self._request._medium, '_path', None) is not None:

                mutter('                  (to %s)', self._request._medium._path)

        self._write_args(args)

        readv_bytes = self._serialise_offsets(body)

        bytes = self._encode_bulk_data(readv_bytes)

            mutter('              %d bytes in readv request', len(readv_bytes))

        self._last_verb = args[0]

 

    def cancel_read_body(self):

        """After expecting a body, a response code may indicate one otherwise.

 

        if 'hpss' in debug.debug_flags:

            if self._request_start_time is not None:

                mutter('   result:   %6.3fs  %s',

                       repr(result)[1:-1])

                self._request_start_time = None

            else:

    def _response_is_unknown_method(self, result_tuple):

        """Raise UnexpectedSmartServerResponse if the response is an 'unknonwn

        method' response to the request.

        :param response: The response from a smart client call_expecting_body

            call.

        :param verb: The verb used in that call.

            # The response will have no body, so we've finished reading.

            self._request.finished_reading()

            raise errors.UnknownSmartMethod(self._last_verb)

    def read_body_bytes(self, count=-1):

        """Read bytes from the body, decoding into a byte stream.

        errors, and then feed the buffer back as read_body_bytes is called.

        """

        if self._body_buffer is not None:

            return self._body_buffer.read(count)

        _body_decoder = LengthPrefixedBodyDecoder()

 

        while not _body_decoder.finished_reading:

            if bytes == '':

                # end of file encountered reading from server

                raise errors.ConnectionReset(

 

    def _recv_tuple(self):

        """Receive a tuple from the medium request."""

 

    def query_version(self):

        """Return protocol version number of the server."""

 

    def _write_protocol_version(self):

        """Write any prefixes this protocol requires.

        Version one doesn't send protocol versions.

        """

 

 

class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):

    """Version two of the client side of the smart protocol.

    This prefixes the request with the value of REQUEST_VERSION_TWO.

    """

 

        if version != self.response_marker:

            self._request.finished_reading()

            raise errors.UnexpectedProtocolVersionMarker(version)

        result = SmartClientRequestProtocolOne._read_response_tuple(self)

        self._response_is_unknown_method(result)

        if response_status == 'success\n':

 

    def _write_protocol_version(self):

        """Write any prefixes this protocol requires.

        Version two sends the value of REQUEST_VERSION_TWO.

        """

        self._request.accept_bytes(self.request_marker)

        """

        # Read no more than 64k at a time so that we don't risk error 10055 (no

        # buffer space available) on Windows.

        _body_decoder = ChunkedBodyDecoder()

        while not _body_decoder.finished_reading:

            if bytes == '':

                # end of file encountered reading from server

                raise errors.ConnectionReset(

 

 

def build_server_protocol_three(backing_transport, write_func,

    request_handler = request.SmartServerRequestHandler(

        backing_transport, commands=request.request_handlers,

    responder = ProtocolThreeResponder(write_func)

    message_handler = message.ConventionalRequestHandler(request_handler, responder)

    return ProtocolThreeDecoder(message_handler)

            # We do *not* set self.decoding_failed here.  The message handler

            # has raised an error, but the decoder is still able to parse bytes

            # and determine when this message ends.

            self.message_handler.protocol_error(exception.exc_value)

            # The state machine is ready to continue decoding, but the

            # exception has interrupted the loop that runs the state machine.

            self.message_handler.protocol_error(exception)

 

    def _extract_length_prefixed_bytes(self):

            # A length prefix by itself is 4 bytes, and we don't even have that

            # many yet.

            raise _NeedMoreBytes(4)

        end_of_bytes = 4 + length

            # We haven't yet read as many bytes as the length-prefix says there

            # are.

            raise _NeedMoreBytes(end_of_bytes)

        # Extract the bytes from the buffer.

        return bytes

 

    def _extract_prefixed_bencoded_data(self):

        prefixed_bytes = self._extract_length_prefixed_bytes()

        try:

        except ValueError:

            raise errors.SmartProtocolError(

                'Bytes %r not bencoded' % (prefixed_bytes,))

        return decoded

 

    def _extract_single_byte(self):

            # The buffer is empty

            raise _NeedMoreBytes(1)

        return one_byte

 

    def _state_accept_expecting_protocol_version(self):

        if needed_bytes > 0:

            # We don't have enough bytes to check if the protocol version

            # marker is right.  But we can check if it is already wrong by

            # len(MESSAGE_VERSION_THREE) bytes.  So if the bytes we have so far

            # are wrong then we should just raise immediately rather than

            # stall.]

                # We have enough bytes to know the protocol version is wrong

            raise _NeedMoreBytes(len(MESSAGE_VERSION_THREE))

        self.state_accept = self._state_accept_expecting_headers

 

    def _state_accept_expecting_headers(self):

            self.message_handler.headers_received(decoded)

        except:

            raise errors.SmartMessageHandlerError(sys.exc_info())

    def _state_accept_expecting_message_part(self):

        message_part_kind = self._extract_single_byte()

        if message_part_kind == 'o':

            raise errors.SmartMessageHandlerError(sys.exc_info())

 

    def done(self):

        self.state_accept = self._state_accept_reading_unused

        try:

            self.message_handler.end_received()

            raise errors.SmartMessageHandlerError(sys.exc_info())

 

    def _state_accept_reading_unused(self):

 

    def next_read_size(self):

        if self.state_accept == self._state_accept_reading_unused:

            return 0

        else:

            if self._number_needed_bytes is not None:

            else:

                raise AssertionError("don't know how many bytes are expected!")

 

class _ProtocolThreeEncoder(object):

 

    response_marker = request_marker = MESSAGE_VERSION_THREE

 

    def __init__(self, write_func):

        self._real_write_func = write_func

 

    def _write_func(self, bytes):

 

    def flush(self):

        if self._buf:

 

    def _serialise_offsets(self, offsets):

        """Serialise a readv offset list."""

        for start, length in offsets:

            txt.append('%d,%d' % (start, length))

        return '\n'.join(txt)

    def _write_protocol_version(self):

        self._write_func(MESSAGE_VERSION_THREE)