( )⚙ D2907 wireproto: add streams to frame-based protocol

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2907

wireproto: add streams to frame-based protocol
ClosedPublic

Authored by indygreg on Mar 20 2018, 5:41 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG9bfcbe4f4745: wireproto: add streams to frame-based protocol
Summary

Previously, the frame-based protocol was just a series of frames,
with each frame associated with a request ID.

In order to scale the protocol, we'll want to enable the use of
compression. While it is possible to enable compression at the
socket/pipe level, this has its disadvantages. The big one is it
undermines the point of frames being standalone, atomic units that
can be read and written: if you add compression above the framing
protocol, you are back to having a stream-based protocol as opposed
to something frame-based.

So in order to preserve frames, compression needs to occur at
the frame payload level.

Compressing each frame's payload individually will limit compression
ratios because the window size of the compressor will be limited
by the max frame size, which is 32-64kb as currently defined. It
will also add CPU overhead, as it is more efficient for compressors
to operate on fewer, larger blocks of data than more, smaller blocks.

So compressing each frame independently is out.

This means we need to compress each frame's payload as if it is part
of a larger stream.

The simplest approach is to have 1 stream per connection. This
could certainly work. However, it has disadvantages (documented below).

We could also have 1 stream per RPC/command invocation. (This is the
model HTTP/2 goes with.) This also has disadvantages.

The main disadvantage to one global stream is that it has the very
real potential to create CPU bottlenecks doing compression. Networks
are only getting faster and the performance of single CPU cores has
been relatively flat. Newer compression formats like zstandard offer
better CPU cycle efficiency than predecessors like zlib. But it still
all too common to saturate your CPU with compression overhead long
before you saturate the network pipe.

The main disadvantage with streams per request is that you can't
reap the benefits of the compression context for multiple requests.
For example, if you send 1000 RPC requests (or HTTP/2 requests for
that matter), the response to each would have its own compression
context. The overall size of the raw responses would be larger because
compression contexts wouldn't be able to reference data from another
request or response.

The approach for streams as implemented in this commit is to support
N streams per connection and for streams to potentially span requests
and responses. As explained by the added internals docs, this
facilitates servers and clients delegating independent streams and
compression to independent threads / CPU cores. This helps alleviate
the CPU bottleneck of compression. This design also allows compression
contexts to be reused across requests/responses. This can result in
improved compression ratios and less overhead for compressors and
decompressors having to build new contexts.

Another feature that was defined was the ability for individual frames
within a stream to declare whether that individual frame's payload
uses the content encoding (read: compression) defined by the stream.
The idea here is that some servers may serve data from a combination
of caches and dynamic resolution. Data coming from caches may be
pre-compressed. We want to facilitate servers being able to essentially
stream bytes from caches to the wire with minimal overhead. Being
able to mix and match with frames are compressed within a stream
enables these types of advanced server functionality.

This commit defines the new streams mechanism. Basic code for
supporting streams in frames has been added. But that code is
seriously lacking and doesn't fully conform to the defined protocol.
For example, we don't close any streams. And support for content
encoding within streams is not yet implemented. The change was
rather invasive and I didn't think it would be reasonable to implement
the entire feature in a single commit.

For the record, I would have loved to reuse an existing multiplexing
protocol to build the new wire protocol on top of. However, I couldn't
find a protocol that offers the performance and scaling characteristics
that I desired. Namely, it should support multiple compression
contexts to facilitate scaling out to multiple CPU cores and
compression contexts should be able to live longer than single RPC
requests. HTTP/2 *almost* fits the bill. But the semantics of HTTP
message exchange state that streams can only live for a single
request-response. We /could/ tunnel on top of HTTP/2 streams and
frames with HEADER and DATA frames. But there's no guarantee that
HTTP/2 libraries and proxies would allow us to use HTTP/2 streams
and frames without the HTTP message exchange semantics defined in
RFC 7540 Section 8. Other RPC protocols like gRPC tunnel are built
on top of HTTP/2 and thus preserve its semantics of stream per
RPC invocation. Even QUIC does this. We could attempt to invent a
higher-level stream that spans HTTP/2 streams. But this would be
violating HTTP/2 because there is no guarantee that HTTP/2 streams
are routed to the same server. The best we can do - which is what
this protocol does - is shoehorn all request and response data into
a single HTTP message and create streams within. At that point, we've
defined a Content-Type in HTTP parlance. It just so happens our
media type can also work as a standalone, stream-based protocol,
without leaning on HTTP or similar protocol.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Mar 20 2018, 5:41 PM
indygreg updated this revision to Diff 7325.Mar 26 2018, 5:35 PM
indygreg updated this revision to Diff 7435.Mar 30 2018, 4:32 PM
durin42 accepted this revision.Apr 3 2018, 11:28 AM
This revision is now accepted and ready to land.Apr 3 2018, 11:28 AM
Closed by commit rHG9bfcbe4f4745: wireproto: add streams to frame-based protocol (authored by indygreg). · Explain WhyApr 3 2018, 12:49 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7549

mercurial/debugcommands.py

Specifying Unified Frame-Based Protocol Frames Specifying Unified Frame-Based Protocol Frames
---------------------------------------------- ----------------------------------------------
It is possible to emit a *Unified Frame-Based Protocol* by using special It is possible to emit a *Unified Frame-Based Protocol* by using special
syntax. syntax.
A frame is composed as a type, flags, and payload. These can be parsed A frame is composed as a type, flags, and payload. These can be parsed
from a string of the form ``<requestid> <type> <flags> <payload>``. That is, from a string of the form:
4 space-delimited strings.
``payload`` is the simplest: it is evaluated as a Python byte string <request-id> <stream-id> <stream-flags> <type> <flags> <payload>
literal.
``requestid`` is an integer defining the request identifier. ``request-id`` and ``stream-id`` are integers defining the request and
stream identifiers.
``type`` can be an integer value for the frame type or the string name ``type`` can be an integer value for the frame type or the string name
of the type. The strings are defined in ``wireprotoframing.py``. e.g. of the type. The strings are defined in ``wireprotoframing.py``. e.g.
``command-name``. ``command-name``.
``flags`` is a ``|`` delimited list of flag components. Each component ``stream-flags`` and ``flags`` are a ``|`` delimited list of flag
(and there can be just one) can be an integer or a flag name for the components. Each component (and there can be just one) can be an integer
specified frame type. Values are resolved to integers and then bitwise or a flag name for stream flags or frame flags, respectively. Values are
OR'd together. resolved to integers and then bitwise OR'd together.
``payload`` is is evaluated as a Python byte string literal.
""" """
opts = pycompat.byteskwargs(opts) opts = pycompat.byteskwargs(opts)
if opts['localssh'] and not repo: if opts['localssh'] and not repo:
raise error.Abort(_('--localssh requires a repository')) raise error.Abort(_('--localssh requires a repository'))
if opts['peer'] and opts['peer'] not in ('raw', 'ssh1', 'ssh2'): if opts['peer'] and opts['peer'] not in ('raw', 'ssh1', 'ssh2'):
raise error.Abort(_('invalid value for --peer'), raise error.Abort(_('invalid value for --peer'),

mercurial/help/internals/wireprotocol.txt

The *Unified Frame-Based Protocol* is a communications protocol between The *Unified Frame-Based Protocol* is a communications protocol between
Mercurial peers. The protocol aims to be mostly transport agnostic Mercurial peers. The protocol aims to be mostly transport agnostic
(works similarly on HTTP, SSH, etc). (works similarly on HTTP, SSH, etc).
To operate the protocol, a bi-directional, half-duplex pipe supporting To operate the protocol, a bi-directional, half-duplex pipe supporting
ordered sends and receives is required. That is, each peer has one pipe ordered sends and receives is required. That is, each peer has one pipe
for sending data and another for receiving. for sending data and another for receiving.
All data is read and written in atomic units called *frames*. These
are conceptually similar to TCP packets. Higher-level functionality
is built on the exchange and processing of frames.
All frames are associated with a *stream*. A *stream* provides a
unidirectional grouping of frames. Streams facilitate two goals:
content encoding and parallelism. There is a dedicated section on
streams below.
The protocol is request-response based: the client issues requests to The protocol is request-response based: the client issues requests to
the server, which issues replies to those requests. Server-initiated the server, which issues replies to those requests. Server-initiated
messaging is not currently supported, but this specification carves messaging is not currently supported, but this specification carves
out room to implement it. out room to implement it.
All data is read and written in atomic units called *frames*. These
are conceptually similar to TCP packets. Higher-level functionality
is built on the exchange and processing of frames.
All frames are associated with a numbered request. Frames can thus All frames are associated with a numbered request. Frames can thus
be logically grouped by their request ID. be logically grouped by their request ID.
Frames begin with a 6 octet header followed by a variable length Frames begin with an 8 octet header followed by a variable length
payload:: payload::
+-----------------------------------------------+ +------------------------------------------------+
| Length (24) | | Length (24) |
+---------------------------------+-------------+ +--------------------------------+---------------+
| Request ID (16) | | Request ID (16) | Stream ID (8) |
+----------+-----------+----------+ +------------------+-------------+---------------+
| Type (4) | Flags (4) | | Stream Flags (8) |
+==========+===========+========================================| +-----------+------+
| Type (4) |
+-----------+
| Flags (4) |
+===========+===================================================|
| Frame Payload (0...) ... | Frame Payload (0...) ...
+---------------------------------------------------------------+ +---------------------------------------------------------------+
The length of the frame payload is expressed as an unsigned 24 bit The length of the frame payload is expressed as an unsigned 24 bit
little endian integer. Values larger than 65535 MUST NOT be used unless little endian integer. Values larger than 65535 MUST NOT be used unless
given permission by the server as part of the negotiated capabilities given permission by the server as part of the negotiated capabilities
during the handshake. The frame header is not part of the advertised during the handshake. The frame header is not part of the advertised
frame length. frame length. The payload length is the over-the-wire length. If there
is content encoding applied to the payload as part of the frame's stream,
the length is the output of that content encoding, not the input.
The 16-bit ``Request ID`` field denotes the integer request identifier, The 16-bit ``Request ID`` field denotes the integer request identifier,
stored as an unsigned little endian integer. Odd numbered requests are stored as an unsigned little endian integer. Odd numbered requests are
client-initiated. Even numbered requests are server-initiated. This client-initiated. Even numbered requests are server-initiated. This
refers to where the *request* was initiated - not where the *frame* was refers to where the *request* was initiated - not where the *frame* was
initiated, so servers will send frames with odd ``Request ID`` in initiated, so servers will send frames with odd ``Request ID`` in
response to client-initiated requests. Implementations are advised to response to client-initiated requests. Implementations are advised to
start ordering request identifiers at ``1`` and ``0``, increment by start ordering request identifiers at ``1`` and ``0``, increment by
``2``, and wrap around if all available numbers have been exhausted. ``2``, and wrap around if all available numbers have been exhausted.
The 4-bit ``Type`` field denotes the type of message being sent. The 8-bit ``Stream ID`` field denotes the stream that the frame is
associated with. Frames belonging to a stream may have content
encoding applied and the receiver may need to decode the raw frame
payload to obtain the original data. Odd numbered IDs are
client-initiated. Even numbered IDs are server-initiated.
The 8-bit ``Stream Flags`` field defines stream processing semantics.
See the section on streams below.
The 4-bit ``Type`` field denotes the type of frame being sent.
The 4-bit ``Flags`` field defines special, per-type attributes for The 4-bit ``Flags`` field defines special, per-type attributes for
the frame. the frame.
The sections below define the frame types and their behavior. The sections below define the frame types and their behavior.
Command Request (``0x01``) Command Request (``0x01``)
-------------------------- --------------------------
All data to be printed MUST be encoded into a single frame: this frame All data to be printed MUST be encoded into a single frame: this frame
does not support spanning data across multiple frames. does not support spanning data across multiple frames.
All textual data encoded in these frames is assumed to be line delimited. All textual data encoded in these frames is assumed to be line delimited.
The last atom in the frame SHOULD end with a newline (``\n``). If it The last atom in the frame SHOULD end with a newline (``\n``). If it
doesn't, clients MAY add a newline to facilitate immediate printing. doesn't, clients MAY add a newline to facilitate immediate printing.
Stream Encoding Settings (``0x08``)
-----------------------------------
This frame type holds information defining the content encoding
settings for a *stream*.
This frame type is likely consumed by the protocol layer and is not
passed on to applications.
This frame type MUST ONLY occur on frames having the *Beginning of Stream*
``Stream Flag`` set.
The payload of this frame defines what content encoding has (possibly)
been applied to the payloads of subsequent frames in this stream.
The payload begins with an 8-bit integer defining the length of the
encoding *profile*, followed by the string name of that profile, which
must be an ASCII string. All bytes that follow can be used by that
profile for supplemental settings definitions. See the section below
on defined encoding profiles.
Stream States and Flags
-----------------------
Streams can be in two states: *open* and *closed*. An *open* stream
is active and frames attached to that stream could arrive at any time.
A *closed* stream is not active. If a frame attached to a *closed*
stream arrives, that frame MUST have an appropriate stream flag
set indicating beginning of stream. All streams are in the *closed*
state by default.
The ``Stream Flags`` field denotes a set of bit flags for defining
the relationship of this frame within a stream. The following flags
are defined:
0x01
Beginning of stream. The first frame in the stream MUST set this
flag. When received, the ``Stream ID`` this frame is attached to
becomes ``open``.
0x02
End of stream. The last frame in a stream MUST set this flag. When
received, the ``Stream ID`` this frame is attached to becomes
``closed``. Any content encoding context associated with this stream
can be destroyed after processing the payload of this frame.
0x04
Apply content encoding. When set, any content encoding settings
defined by the stream should be applied when attempting to read
the frame. When not set, the frame payload isn't encoded.
Streams
-------
Streams - along with ``Request IDs`` - facilitate grouping of frames.
But the purpose of each is quite different and the groupings they
constitute are independent.
A ``Request ID`` is essentially a tag. It tells you which logical
request a frame is associated with.
A *stream* is a sequence of frames grouped for the express purpose
of applying a stateful encoding or for denoting sub-groups of frames.
Unlike ``Request ID``s which span the request and response, a stream
is unidirectional and stream IDs are independent from client to
server.
There is no strict hierarchical relationship between ``Request IDs``
and *streams*. A stream can contain frames having multiple
``Request IDs``. Frames belonging to the same ``Request ID`` can
span multiple streams.
One goal of streams is to facilitate content encoding. A stream can
define an encoding to be applied to frame payloads. For example, the
payload transmitted over the wire may contain output from a
zstandard compression operation and the receiving end may decompress
that payload to obtain the original data.
The other goal of streams is to facilitate concurrent execution. For
example, a server could spawn 4 threads to service a request that can
be easily parallelized. Each of those 4 threads could write into its
own stream. Those streams could then in turn be delivered to 4 threads
on the receiving end, with each thread consuming its stream in near
isolation. The *main* thread on both ends merely does I/O and
encodes/decodes frame headers: the bulk of the work is done by worker
threads.
In addition, since content encoding is defined per stream, each
*worker thread* could perform potentially CPU bound work concurrently
with other threads. This approach of applying encoding at the
sub-protocol / stream level eliminates a potential resource constraint
on the protocol stream as a whole (it is common for the throughput of
a compression engine to be smaller than the throughput of a network).
Having multiple streams - each with their own encoding settings - also
facilitates the use of advanced data compression techniques. For
example, a transmitter could see that it is generating data faster
and slower than the receiving end is consuming it and adjust its
compression settings to trade CPU for compression ratio accordingly.
While streams can define a content encoding, not all frames within
that stream must use that content encoding. This can be useful when
data is being served from caches and being derived dynamically. A
cache could pre-compressed data so the server doesn't have to
recompress it. The ability to pick and choose which frames are
compressed allows servers to easily send data to the wire without
involving potentially expensive encoding overhead.
Content Encoding Profiles
-------------------------
Streams can have named content encoding *profiles* associated with
them. A profile defines a shared understanding of content encoding
settings and behavior.
The following profiles are defined:
TBD
Issuing Commands Issuing Commands
---------------- ----------------
A client can request that a remote run a command by sending it A client can request that a remote run a command by sending it
frames defining that command. This logical stream is composed of frames defining that command. This logical stream is composed of
1 ``Command Request`` frame, 0 or more ``Command Argument`` frames, 1 ``Command Request`` frame, 0 or more ``Command Argument`` frames,
and 0 or more ``Command Data`` frames. and 0 or more ``Command Data`` frames.

mercurial/wireprotoframing.py

from . import ( from . import (
error, error,
util, util,
) )
from .utils import ( from .utils import (
stringutil, stringutil,
) )
FRAME_HEADER_SIZE = 6 FRAME_HEADER_SIZE = 8
DEFAULT_MAX_FRAME_SIZE = 32768 DEFAULT_MAX_FRAME_SIZE = 32768
STREAM_FLAG_BEGIN_STREAM = 0x01
STREAM_FLAG_END_STREAM = 0x02
STREAM_FLAG_ENCODING_APPLIED = 0x04
STREAM_FLAGS = {
b'stream-begin': STREAM_FLAG_BEGIN_STREAM,
b'stream-end': STREAM_FLAG_END_STREAM,
b'encoded': STREAM_FLAG_ENCODING_APPLIED,
}
FRAME_TYPE_COMMAND_NAME = 0x01 FRAME_TYPE_COMMAND_NAME = 0x01
FRAME_TYPE_COMMAND_ARGUMENT = 0x02 FRAME_TYPE_COMMAND_ARGUMENT = 0x02
FRAME_TYPE_COMMAND_DATA = 0x03 FRAME_TYPE_COMMAND_DATA = 0x03
FRAME_TYPE_BYTES_RESPONSE = 0x04 FRAME_TYPE_BYTES_RESPONSE = 0x04
FRAME_TYPE_ERROR_RESPONSE = 0x05 FRAME_TYPE_ERROR_RESPONSE = 0x05
FRAME_TYPE_TEXT_OUTPUT = 0x06 FRAME_TYPE_TEXT_OUTPUT = 0x06
FRAME_TYPE_STREAM_SETTINGS = 0x08
FRAME_TYPES = { FRAME_TYPES = {
b'command-name': FRAME_TYPE_COMMAND_NAME, b'command-name': FRAME_TYPE_COMMAND_NAME,
b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT, b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT,
b'command-data': FRAME_TYPE_COMMAND_DATA, b'command-data': FRAME_TYPE_COMMAND_DATA,
b'bytes-response': FRAME_TYPE_BYTES_RESPONSE, b'bytes-response': FRAME_TYPE_BYTES_RESPONSE,
b'error-response': FRAME_TYPE_ERROR_RESPONSE, b'error-response': FRAME_TYPE_ERROR_RESPONSE,
b'text-output': FRAME_TYPE_TEXT_OUTPUT, b'text-output': FRAME_TYPE_TEXT_OUTPUT,
b'stream-settings': FRAME_TYPE_STREAM_SETTINGS,
} }
FLAG_COMMAND_NAME_EOS = 0x01 FLAG_COMMAND_NAME_EOS = 0x01
FLAG_COMMAND_NAME_HAVE_ARGS = 0x02 FLAG_COMMAND_NAME_HAVE_ARGS = 0x02
FLAG_COMMAND_NAME_HAVE_DATA = 0x04 FLAG_COMMAND_NAME_HAVE_DATA = 0x04
FLAGS_COMMAND = { FLAGS_COMMAND = {
b'eos': FLAG_COMMAND_NAME_EOS, b'eos': FLAG_COMMAND_NAME_EOS,
# Maps frame types to their available flags. # Maps frame types to their available flags.
FRAME_TYPE_FLAGS = { FRAME_TYPE_FLAGS = {
FRAME_TYPE_COMMAND_NAME: FLAGS_COMMAND, FRAME_TYPE_COMMAND_NAME: FLAGS_COMMAND,
FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT, FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT,
FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA, FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE, FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE,
FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE, FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE,
FRAME_TYPE_TEXT_OUTPUT: {}, FRAME_TYPE_TEXT_OUTPUT: {},
FRAME_TYPE_STREAM_SETTINGS: {},
} }
ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH') ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH')
@attr.s(slots=True) @attr.s(slots=True)
class frameheader(object): class frameheader(object):
"""Represents the data in a frame header.""" """Represents the data in a frame header."""
length = attr.ib() length = attr.ib()
requestid = attr.ib() requestid = attr.ib()
streamid = attr.ib()
streamflags = attr.ib()
typeid = attr.ib() typeid = attr.ib()
flags = attr.ib() flags = attr.ib()
@attr.s(slots=True) @attr.s(slots=True)
class frame(object): class frame(object):
"""Represents a parsed frame.""" """Represents a parsed frame."""
requestid = attr.ib() requestid = attr.ib()
streamid = attr.ib()
streamflags = attr.ib()
typeid = attr.ib() typeid = attr.ib()
flags = attr.ib() flags = attr.ib()
payload = attr.ib() payload = attr.ib()
def makeframe(requestid, typeid, flags, payload): def makeframe(requestid, streamid, streamflags, typeid, flags, payload):
"""Assemble a frame into a byte array.""" """Assemble a frame into a byte array."""
# TODO assert size of payload. # TODO assert size of payload.
frame = bytearray(FRAME_HEADER_SIZE + len(payload)) frame = bytearray(FRAME_HEADER_SIZE + len(payload))
# 24 bits length # 24 bits length
# 16 bits request id # 16 bits request id
# 8 bits stream id
# 8 bits stream flags
# 4 bits type # 4 bits type
# 4 bits flags # 4 bits flags
l = struct.pack(r'<I', len(payload)) l = struct.pack(r'<I', len(payload))
frame[0:3] = l[0:3] frame[0:3] = l[0:3]
struct.pack_into(r'<H', frame, 3, requestid) struct.pack_into(r'<HBB', frame, 3, requestid, streamid, streamflags)
frame[5] = (typeid << 4) | flags frame[7] = (typeid << 4) | flags
frame[6:] = payload frame[8:] = payload
return frame return frame
def makeframefromhumanstring(s): def makeframefromhumanstring(s):
"""Create a frame from a human readable string """Create a frame from a human readable string
Strings have the form: Strings have the form:
<request-id> <type> <flags> <payload> <request-id> <stream-id> <stream-flags> <type> <flags> <payload>
This can be used by user-facing applications and tests for creating This can be used by user-facing applications and tests for creating
frames easily without having to type out a bunch of constants. frames easily without having to type out a bunch of constants.
Request ID is an integer. Request ID and stream IDs are integers.
Frame type and flags can be specified by integer or named constant. Stream flags, frame type, and flags can be specified by integer or
named constant.
Flags can be delimited by `|` to bitwise OR them together. Flags can be delimited by `|` to bitwise OR them together.
""" """
requestid, frametype, frameflags, payload = s.split(b' ', 3) fields = s.split(b' ', 5)
requestid, streamid, streamflags, frametype, frameflags, payload = fields
requestid = int(requestid) requestid = int(requestid)
streamid = int(streamid)
finalstreamflags = 0
for flag in streamflags.split(b'|'):
if flag in STREAM_FLAGS:
finalstreamflags |= STREAM_FLAGS[flag]
else:
finalstreamflags |= int(flag)
if frametype in FRAME_TYPES: if frametype in FRAME_TYPES:
frametype = FRAME_TYPES[frametype] frametype = FRAME_TYPES[frametype]
else: else:
frametype = int(frametype) frametype = int(frametype)
finalflags = 0 finalflags = 0
validflags = FRAME_TYPE_FLAGS[frametype] validflags = FRAME_TYPE_FLAGS[frametype]
for flag in frameflags.split(b'|'): for flag in frameflags.split(b'|'):
if flag in validflags: if flag in validflags:
finalflags |= validflags[flag] finalflags |= validflags[flag]
else: else:
finalflags |= int(flag) finalflags |= int(flag)
payload = stringutil.unescapestr(payload) payload = stringutil.unescapestr(payload)
return makeframe(requestid=requestid, typeid=frametype, return makeframe(requestid=requestid, streamid=streamid,
streamflags=finalstreamflags, typeid=frametype,
flags=finalflags, payload=payload) flags=finalflags, payload=payload)
def parseheader(data): def parseheader(data):
"""Parse a unified framing protocol frame header from a buffer. """Parse a unified framing protocol frame header from a buffer.
The header is expected to be in the buffer at offset 0 and the The header is expected to be in the buffer at offset 0 and the
buffer is expected to be large enough to hold a full header. buffer is expected to be large enough to hold a full header.
""" """
# 24 bits payload length (little endian) # 24 bits payload length (little endian)
# 16 bits request ID
# 8 bits stream ID
# 8 bits stream flags
# 4 bits frame type # 4 bits frame type
# 4 bits frame flags # 4 bits frame flags
# ... payload # ... payload
framelength = data[0] + 256 * data[1] + 16384 * data[2] framelength = data[0] + 256 * data[1] + 16384 * data[2]
requestid = struct.unpack_from(r'<H', data, 3)[0] requestid, streamid, streamflags = struct.unpack_from(r'<HBB', data, 3)
typeflags = data[5] typeflags = data[7]
frametype = (typeflags & 0xf0) >> 4 frametype = (typeflags & 0xf0) >> 4
frameflags = typeflags & 0x0f frameflags = typeflags & 0x0f
return frameheader(framelength, requestid, frametype, frameflags) return frameheader(framelength, requestid, streamid, streamflags,
frametype, frameflags)
def readframe(fh): def readframe(fh):
"""Read a unified framing protocol frame from a file object. """Read a unified framing protocol frame from a file object.
Returns a 3-tuple of (type, flags, payload) for the decoded frame or Returns a 3-tuple of (type, flags, payload) for the decoded frame or
None if no frame is available. May raise if a malformed frame is None if no frame is available. May raise if a malformed frame is
seen. seen.
""" """
h = parseheader(header) h = parseheader(header)
payload = fh.read(h.length) payload = fh.read(h.length)
if len(payload) != h.length: if len(payload) != h.length:
raise error.Abort(_('frame length error: expected %d; got %d') % raise error.Abort(_('frame length error: expected %d; got %d') %
(h.length, len(payload))) (h.length, len(payload)))
return frame(h.requestid, h.typeid, h.flags, payload) return frame(h.requestid, h.streamid, h.streamflags, h.typeid, h.flags,
payload)
def createcommandframes(stream, requestid, cmd, args, datafh=None): def createcommandframes(stream, requestid, cmd, args, datafh=None):
"""Create frames necessary to transmit a request to run a command. """Create frames necessary to transmit a request to run a command.
This is a generator of bytearrays. Each item represents a frame This is a generator of bytearrays. Each item represents a frame
ready to be sent over the wire to a peer. ready to be sent over the wire to a peer.
""" """
flags = 0 flags = 0
yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_TEXT_OUTPUT, typeid=FRAME_TYPE_TEXT_OUTPUT,
flags=0, flags=0,
payload=b''.join(atomchunks)) payload=b''.join(atomchunks))
class stream(object): class stream(object):
"""Represents a logical unidirectional series of frames.""" """Represents a logical unidirectional series of frames."""
def __init__(self, streamid, active=False):
self.streamid = streamid
self._active = False
def makeframe(self, requestid, typeid, flags, payload): def makeframe(self, requestid, typeid, flags, payload):
"""Create a frame to be sent out over this stream. """Create a frame to be sent out over this stream.
Only returns the frame instance. Does not actually send it. Only returns the frame instance. Does not actually send it.
""" """
return makeframe(requestid, typeid, flags, payload) streamflags = 0
if not self._active:
streamflags |= STREAM_FLAG_BEGIN_STREAM
self._active = True
return makeframe(requestid, self.streamid, streamflags, typeid, flags,
payload)
def ensureserverstream(stream):
if stream.streamid % 2:
raise error.ProgrammingError('server should only write to even '
'numbered streams; %d is not even' %
stream.streamid)
class serverreactor(object): class serverreactor(object):
"""Holds state of a server handling frame-based protocol requests. """Holds state of a server handling frame-based protocol requests.
This class is the "brain" of the unified frame-based protocol server This class is the "brain" of the unified frame-based protocol server
component. While the protocol is stateless from the perspective of component. While the protocol is stateless from the perspective of
requests/commands, something needs to track which frames have been requests/commands, something needs to track which frames have been
received, what frames to expect, etc. This class is that thing. received, what frames to expect, etc. This class is that thing.
events that would normally generate output frames (such as a command events that would normally generate output frames (such as a command
response being ready) will instead defer instructing the consumer to response being ready) will instead defer instructing the consumer to
send those frames. This is useful for half-duplex transports where the send those frames. This is useful for half-duplex transports where the
sender cannot receive until all data has been transmitted. sender cannot receive until all data has been transmitted.
""" """
self._deferoutput = deferoutput self._deferoutput = deferoutput
self._state = 'idle' self._state = 'idle'
self._bufferedframegens = [] self._bufferedframegens = []
# stream id -> stream instance for all active streams from the client.
self._incomingstreams = {}
# request id -> dict of commands that are actively being received. # request id -> dict of commands that are actively being received.
self._receivingcommands = {} self._receivingcommands = {}
# Request IDs that have been received and are actively being processed. # Request IDs that have been received and are actively being processed.
# Once all output for a request has been sent, it is removed from this # Once all output for a request has been sent, it is removed from this
# set. # set.
self._activecommands = set() self._activecommands = set()
def onframerecv(self, frame): def onframerecv(self, frame):
"""Process a frame that has been received off the wire. """Process a frame that has been received off the wire.
Returns a dict with an ``action`` key that details what action, Returns a dict with an ``action`` key that details what action,
if any, the consumer should take next. if any, the consumer should take next.
""" """
if not frame.streamid % 2:
self._state = 'errored'
return self._makeerrorresult(
_('received frame with even numbered stream ID: %d') %
frame.streamid)
if frame.streamid not in self._incomingstreams:
if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
self._state = 'errored'
return self._makeerrorresult(
_('received frame on unknown inactive stream without '
'beginning of stream flag set'))
self._incomingstreams[frame.streamid] = stream(frame.streamid)
if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
# TODO handle decoding frames
self._state = 'errored'
raise error.ProgrammingError('support for decoding stream payloads '
'not yet implemented')
if frame.streamflags & STREAM_FLAG_END_STREAM:
del self._incomingstreams[frame.streamid]
handlers = { handlers = {
'idle': self._onframeidle, 'idle': self._onframeidle,
'command-receiving': self._onframecommandreceiving, 'command-receiving': self._onframecommandreceiving,
'errored': self._onframeerrored, 'errored': self._onframeerrored,
} }
meth = handlers.get(self._state) meth = handlers.get(self._state)
if not meth: if not meth:
raise error.ProgrammingError('unhandled state: %s' % self._state) raise error.ProgrammingError('unhandled state: %s' % self._state)
return meth(frame) return meth(frame)
def onbytesresponseready(self, stream, requestid, data): def onbytesresponseready(self, stream, requestid, data):
"""Signal that a bytes response is ready to be sent to the client. """Signal that a bytes response is ready to be sent to the client.
The raw bytes response is passed as an argument. The raw bytes response is passed as an argument.
""" """
ensureserverstream(stream)
def sendframes(): def sendframes():
for frame in createbytesresponseframesfrombytes(stream, requestid, for frame in createbytesresponseframesfrombytes(stream, requestid,
data): data):
yield frame yield frame
self._activecommands.remove(requestid) self._activecommands.remove(requestid)
result = sendframes() result = sendframes()
for frame in gen: for frame in gen:
yield frame yield frame
return 'sendframes', { return 'sendframes', {
'framegen': makegen(), 'framegen': makegen(),
} }
def onapplicationerror(self, stream, requestid, msg): def onapplicationerror(self, stream, requestid, msg):
ensureserverstream(stream)
return 'sendframes', { return 'sendframes', {
'framegen': createerrorframe(stream, requestid, msg, 'framegen': createerrorframe(stream, requestid, msg,
application=True), application=True),
} }
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,

mercurial/wireprotoserver.py

res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(_('command in frame must match command in URL')) res.setbodybytes(_('command in frame must match command in URL'))
return True return True
rsp = wireproto.dispatch(repo, proto, command['command']) rsp = wireproto.dispatch(repo, proto, command['command'])
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
stream = wireprotoframing.stream() stream = wireprotoframing.stream(2)
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
action, meta = reactor.onbytesresponseready(stream, action, meta = reactor.onbytesresponseready(stream,
command['requestid'], command['requestid'],
rsp.data) rsp.data)
else: else:
action, meta = reactor.onapplicationerror( action, meta = reactor.onapplicationerror(
_('unhandled response type from wire proto command')) _('unhandled response type from wire proto command'))

tests/test-http-api-httpv2.t

Request to read-only command works out of the box Request to read-only command works out of the box
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name eos customreadonly > frame 1 1 stream-begin command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> *\r\n (glob) s> *\r\n (glob)
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x01\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x01\x01\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 23\r\n s> 25\r\n
s> \x1d\x00\x00\x01\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00\x02\x01Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Request to read-write command fails because server is read-only by default Request to read-write command fails because server is read-only by default
GET to read-write request yields 405 GET to read-write request yields 405
Authorized request for valid read-write command works Authorized request for valid read-write command works
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/rw/customreadonly > httprequest POST api/$HTTPV2/rw/customreadonly
> user-agent: test > user-agent: test
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> frame 1 command-name eos customreadonly > frame 1 1 stream-begin command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 20\r\n s> content-length: 22\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x01\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x01\x01\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 23\r\n s> 25\r\n
s> \x1d\x00\x00\x01\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00\x02\x01Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Authorized request for unknown command is rejected Authorized request for unknown command is rejected
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/rw/badcommand > httprequest POST api/$HTTPV2/rw/badcommand
Command frames can be reflected via debugreflect Command frames can be reflected via debugreflect
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/debugreflect > httprequest POST api/$HTTPV2/ro/debugreflect
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name have-args command1 > frame 1 1 stream-begin command-name have-args command1
> frame 1 command-argument 0 \x03\x00\x04\x00fooval1 > frame 1 1 0 command-argument 0 \x03\x00\x04\x00fooval1
> frame 1 command-argument eoa \x04\x00\x03\x00bar1val > frame 1 1 0 command-argument eoa \x04\x00\x03\x00bar1val
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 48\r\n s> content-length: 54\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x08\x00\x00\x01\x00\x12command1\x0b\x00\x00\x01\x00 \x03\x00\x04\x00fooval1\x0b\x00\x00\x01\x00"\x04\x00\x03\x00bar1val s> \x08\x00\x00\x01\x00\x01\x01\x12command1\x0b\x00\x00\x01\x00\x01\x00 \x03\x00\x04\x00fooval1\x0b\x00\x00\x01\x00\x01\x00"\x04\x00\x03\x00bar1val
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: text/plain\r\n
s> Content-Length: 322\r\n s> Content-Length: 322\r\n
s> \r\n s> \r\n
s> received: 1 2 1 command1\n s> received: 1 2 1 command1\n
s> ["wantframe", {"state": "command-receiving"}]\n s> ["wantframe", {"state": "command-receiving"}]\n
s> received: 2 0 1 \x03\x00\x04\x00fooval1\n s> received: 2 0 1 \x03\x00\x04\x00fooval1\n
s> ["wantframe", {"state": "command-receiving"}]\n s> ["wantframe", {"state": "command-receiving"}]\n
s> received: 2 2 1 \x04\x00\x03\x00bar1val\n s> received: 2 2 1 \x04\x00\x03\x00bar1val\n
s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n
s> received: <no frame>\n s> received: <no frame>\n
s> {"action": "noop"} s> {"action": "noop"}
Multiple requests to regular command URL are not allowed Multiple requests to regular command URL are not allowed
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name eos customreadonly > frame 1 1 stream-begin command-name eos customreadonly
> frame 3 command-name eos customreadonly > frame 3 1 0 command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 40\r\n s> content-length: 44\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x01\x00\x11customreadonly\x0e\x00\x00\x03\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x01\x01\x11customreadonly\x0e\x00\x00\x03\x00\x01\x00\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: text/plain\r\n
s> Content-Length: 46\r\n s> Content-Length: 46\r\n
s> \r\n s> \r\n
s> multiple commands cannot be issued to this URL s> multiple commands cannot be issued to this URL
Multiple requests to "multirequest" URL are allowed Multiple requests to "multirequest" URL are allowed
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/multirequest > httprequest POST api/$HTTPV2/ro/multirequest
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name eos customreadonly > frame 1 1 stream-begin command-name eos customreadonly
> frame 3 command-name eos customreadonly > frame 3 1 0 command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> *\r\n (glob) s> *\r\n (glob)
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x01\x00\x11customreadonly\x0e\x00\x00\x03\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x01\x01\x11customreadonly\x0e\x00\x00\x03\x00\x01\x00\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> *\r\n (glob) s> *\r\n (glob)
s> \x1d\x00\x00\x01\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00\x02\x01Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 23\r\n s> 25\r\n
s> \x1d\x00\x00\x03\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x03\x00\x02\x01Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Interleaved requests to "multirequest" are processed Interleaved requests to "multirequest" are processed
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/multirequest > httprequest POST api/$HTTPV2/ro/multirequest
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name have-args listkeys > frame 1 1 stream-begin command-name have-args listkeys
> frame 3 command-name have-args listkeys > frame 3 1 0 command-name have-args listkeys
> frame 3 command-argument eoa \x09\x00\x09\x00namespacebookmarks > frame 3 1 0 command-argument eoa \x09\x00\x09\x00namespacebookmarks
> frame 1 command-argument eoa \x09\x00\x0a\x00namespacenamespaces > frame 1 1 0 command-argument eoa \x09\x00\x0a\x00namespacenamespaces
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 85\r\n s> content-length: 93\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x08\x00\x00\x01\x00\x12listkeys\x08\x00\x00\x03\x00\x12listkeys\x16\x00\x00\x03\x00" \x00 \x00namespacebookmarks\x17\x00\x00\x01\x00" \x00\n s> \x08\x00\x00\x01\x00\x01\x01\x12listkeys\x08\x00\x00\x03\x00\x01\x00\x12listkeys\x16\x00\x00\x03\x00\x01\x00" \x00 \x00namespacebookmarks\x17\x00\x00\x01\x00\x01\x00" \x00\n
s> \x00namespacenamespaces s> \x00namespacenamespaces
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 6\r\n s> 8\r\n
s> \x00\x00\x00\x03\x00B s> \x00\x00\x00\x03\x00\x02\x01B
s> \r\n s> \r\n
s> 24\r\n s> 26\r\n
s> \x1e\x00\x00\x01\x00Bbookmarks \n s> \x1e\x00\x00\x01\x00\x02\x01Bbookmarks \n
s> namespaces \n s> namespaces \n
s> phases s> phases
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Restart server to disable read-write access Restart server to disable read-write access
Attempting to run a read-write command via multirequest on read-only URL is not allowed Attempting to run a read-write command via multirequest on read-only URL is not allowed
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/multirequest > httprequest POST api/$HTTPV2/ro/multirequest
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame 1 command-name eos unbundle > frame 1 1 stream-begin command-name eos unbundle
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/multirequest HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0002\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0002\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 14\r\n s> content-length: 16\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x08\x00\x00\x01\x00\x11unbundle s> \x08\x00\x00\x01\x00\x01\x01\x11unbundle
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 403 Forbidden\r\n s> HTTP/1.1 403 Forbidden\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: text/plain\r\n
s> Content-Length: 53\r\n s> Content-Length: 53\r\n
s> \r\n s> \r\n
s> insufficient permissions to execute command: unbundle s> insufficient permissions to execute command: unbundle
$ cat error.log $ cat error.log

tests/test-wireproto-serverreactor.py

Emits a generator of results from ``onframerecv()`` calls. Emits a generator of results from ``onframerecv()`` calls.
""" """
for frame in gen: for frame in gen:
header = framing.parseheader(frame) header = framing.parseheader(frame)
payload = frame[framing.FRAME_HEADER_SIZE:] payload = frame[framing.FRAME_HEADER_SIZE:]
assert len(payload) == header.length assert len(payload) == header.length
yield reactor.onframerecv(framing.frame(header.requestid, yield reactor.onframerecv(framing.frame(header.requestid,
header.streamid,
header.streamflags,
header.typeid, header.typeid,
header.flags, header.flags,
payload)) payload))
def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None): def sendcommandframes(reactor, stream, rid, cmd, args, datafh=None):
"""Generate frames to run a command and send them to a reactor.""" """Generate frames to run a command and send them to a reactor."""
return sendframes(reactor, return sendframes(reactor,
framing.createcommandframes(stream, rid, cmd, args, framing.createcommandframes(stream, rid, cmd, args,
datafh)) datafh))
class FrameTests(unittest.TestCase): class FrameTests(unittest.TestCase):
def testdataexactframesize(self): def testdataexactframesize(self):
data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE) data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE)
stream = framing.stream() stream = framing.stream(1)
frames = list(framing.createcommandframes(stream, 1, b'command', frames = list(framing.createcommandframes(stream, 1, b'command',
{}, data)) {}, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-data command'), ffs(b'1 1 stream-begin command-name have-data command'),
ffs(b'1 command-data continuation %s' % data.getvalue()), ffs(b'1 1 0 command-data continuation %s' % data.getvalue()),
ffs(b'1 command-data eos ') ffs(b'1 1 0 command-data eos ')
]) ])
def testdatamultipleframes(self): def testdatamultipleframes(self):
data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1)) data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1))
stream = framing.stream() stream = framing.stream(1)
frames = list(framing.createcommandframes(stream, 1, b'command', {}, frames = list(framing.createcommandframes(stream, 1, b'command', {},
data)) data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-data command'), ffs(b'1 1 stream-begin command-name have-data command'),
ffs(b'1 command-data continuation %s' % ( ffs(b'1 1 0 command-data continuation %s' % (
b'x' * framing.DEFAULT_MAX_FRAME_SIZE)), b'x' * framing.DEFAULT_MAX_FRAME_SIZE)),
ffs(b'1 command-data eos x'), ffs(b'1 1 0 command-data eos x'),
]) ])
def testargsanddata(self): def testargsanddata(self):
data = util.bytesio(b'x' * 100) data = util.bytesio(b'x' * 100)
stream = framing.stream() stream = framing.stream(1)
frames = list(framing.createcommandframes(stream, 1, b'command', { frames = list(framing.createcommandframes(stream, 1, b'command', {
b'key1': b'key1value', b'key1': b'key1value',
b'key2': b'key2value', b'key2': b'key2value',
b'key3': b'key3value', b'key3': b'key3value',
}, data)) }, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'1 command-name have-args|have-data command'), ffs(b'1 1 stream-begin command-name have-args|have-data command'),
ffs(br'1 command-argument 0 \x04\x00\x09\x00key1key1value'), ffs(br'1 1 0 command-argument 0 \x04\x00\x09\x00key1key1value'),
ffs(br'1 command-argument 0 \x04\x00\x09\x00key2key2value'), ffs(br'1 1 0 command-argument 0 \x04\x00\x09\x00key2key2value'),
ffs(br'1 command-argument eoa \x04\x00\x09\x00key3key3value'), ffs(br'1 1 0 command-argument eoa \x04\x00\x09\x00key3key3value'),
ffs(b'1 command-data eos %s' % data.getvalue()), ffs(b'1 1 0 command-data eos %s' % data.getvalue()),
]) ])
def testtextoutputexcessiveargs(self): def testtextoutputexcessiveargs(self):
"""At most 255 formatting arguments are allowed.""" """At most 255 formatting arguments are allowed."""
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'cannot use more than 255 formatting'): 'cannot use more than 255 formatting'):
args = [b'x' for i in range(256)] args = [b'x' for i in range(256)]
list(framing.createtextoutputframe(None, 1, list(framing.createtextoutputframe(None, 1,
def testtextoutputtoolonglabelstring(self): def testtextoutputtoolonglabelstring(self):
with self.assertRaisesRegexp(ValueError, with self.assertRaisesRegexp(ValueError,
'label string cannot be longer than'): 'label string cannot be longer than'):
list(framing.createtextoutputframe(None, 1, [ list(framing.createtextoutputframe(None, 1, [
(b'bleh', [], [b'x' * 65536])])) (b'bleh', [], [b'x' * 65536])]))
def testtextoutput1simpleatom(self): def testtextoutput1simpleatom(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], [])])) (b'foo', [], [])]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x00\x00foo'), ffs(br'1 1 stream-begin text-output 0 \x03\x00\x00\x00foo'),
]) ])
def testtextoutput2simpleatoms(self): def testtextoutput2simpleatoms(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], []), (b'foo', [], []),
(b'bar', [], []), (b'bar', [], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x00\x00foo\x03\x00\x00\x00bar'), ffs(br'1 1 stream-begin text-output 0 '
br'\x03\x00\x00\x00foo\x03\x00\x00\x00bar'),
]) ])
def testtextoutput1arg(self): def testtextoutput1arg(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s', [b'val1'], []), (b'foo %s', [b'val1'], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x06\x00\x00\x01\x04\x00foo %sval1'), ffs(br'1 1 stream-begin text-output 0 '
br'\x06\x00\x00\x01\x04\x00foo %sval1'),
]) ])
def testtextoutput2arg(self): def testtextoutput2arg(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s %s', [b'val', b'value'], []), (b'foo %s %s', [b'val', b'value'], []),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x09\x00\x00\x02\x03\x00\x05\x00' ffs(br'1 1 stream-begin text-output 0 '
br'foo %s %svalvalue'), br'\x09\x00\x00\x02\x03\x00\x05\x00foo %s %svalvalue'),
]) ])
def testtextoutput1label(self): def testtextoutput1label(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo', [], [b'label']), (b'foo', [], [b'label']),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x03\x00\x01\x00\x05foolabel'), ffs(br'1 1 stream-begin text-output 0 '
br'\x03\x00\x01\x00\x05foolabel'),
]) ])
def testargandlabel(self): def testargandlabel(self):
stream = framing.stream() stream = framing.stream(1)
val = list(framing.createtextoutputframe(stream, 1, [ val = list(framing.createtextoutputframe(stream, 1, [
(b'foo %s', [b'arg'], [b'label']), (b'foo %s', [b'arg'], [b'label']),
])) ]))
self.assertEqual(val, [ self.assertEqual(val, [
ffs(br'1 text-output 0 \x06\x00\x01\x01\x05\x03\x00foo %slabelarg'), ffs(br'1 1 stream-begin text-output 0 '
br'\x06\x00\x01\x01\x05\x03\x00foo %slabelarg'),
]) ])
class ServerReactorTests(unittest.TestCase): class ServerReactorTests(unittest.TestCase):
def _sendsingleframe(self, reactor, f): def _sendsingleframe(self, reactor, f):
results = list(sendframes(reactor, [f])) results = list(sendframes(reactor, [f]))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
return results[0] return results[0]
def assertaction(self, res, expected): def assertaction(self, res, expected):
self.assertIsInstance(res, tuple) self.assertIsInstance(res, tuple)
self.assertEqual(len(res), 2) self.assertEqual(len(res), 2)
self.assertIsInstance(res[1], dict) self.assertIsInstance(res[1], dict)
self.assertEqual(res[0], expected) self.assertEqual(res[0], expected)
def assertframesequal(self, frames, framestrings): def assertframesequal(self, frames, framestrings):
expected = [ffs(s) for s in framestrings] expected = [ffs(s) for s in framestrings]
self.assertEqual(list(frames), expected) self.assertEqual(list(frames), expected)
def test1framecommand(self): def test1framecommand(self):
"""Receiving a command in a single frame yields request to run it.""" """Receiving a command in a single frame yields request to run it."""
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {})) results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': None, 'data': None,
}) })
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
def test1argument(self): def test1argument(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 41, b'mycommand', results = list(sendcommandframes(reactor, stream, 41, b'mycommand',
{b'foo': b'bar'})) {b'foo': b'bar'}))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 41, 'requestid': 41,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar'}, 'args': {b'foo': b'bar'},
'data': None, 'data': None,
}) })
def testmultiarguments(self): def testmultiarguments(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', results = list(sendcommandframes(reactor, stream, 1, b'mycommand',
{b'foo': b'bar', b'biz': b'baz'})) {b'foo': b'bar', b'biz': b'baz'}))
self.assertEqual(len(results), 3) self.assertEqual(len(results), 3)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'wantframe') self.assertaction(results[1], 'wantframe')
self.assertaction(results[2], 'runcommand') self.assertaction(results[2], 'runcommand')
self.assertEqual(results[2][1], { self.assertEqual(results[2][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar', b'biz': b'baz'}, 'args': {b'foo': b'bar', b'biz': b'baz'},
'data': None, 'data': None,
}) })
def testsimplecommanddata(self): def testsimplecommanddata(self):
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}, results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {},
util.bytesio(b'data!'))) util.bytesio(b'data!')))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': b'data!', 'data': b'data!',
}) })
def testmultipledataframes(self): def testmultipledataframes(self):
frames = [ frames = [
ffs(b'1 command-name have-data mycommand'), ffs(b'1 1 stream-begin command-name have-data mycommand'),
ffs(b'1 command-data continuation data1'), ffs(b'1 1 0 command-data continuation data1'),
ffs(b'1 command-data continuation data2'), ffs(b'1 1 0 command-data continuation data2'),
ffs(b'1 command-data eos data3'), ffs(b'1 1 0 command-data eos data3'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertEqual(len(results), 4) self.assertEqual(len(results), 4)
for i in range(3): for i in range(3):
self.assertaction(results[i], 'wantframe') self.assertaction(results[i], 'wantframe')
self.assertaction(results[3], 'runcommand') self.assertaction(results[3], 'runcommand')
self.assertEqual(results[3][1], { self.assertEqual(results[3][1], {
'requestid': 1, 'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': b'data1data2data3', 'data': b'data1data2data3',
}) })
def testargumentanddata(self): def testargumentanddata(self):
frames = [ frames = [
ffs(b'1 command-name have-args|have-data command'), ffs(b'1 1 stream-begin command-name have-args|have-data command'),
ffs(br'1 command-argument 0 \x03\x00\x03\x00keyval'), ffs(br'1 1 0 command-argument 0 \x03\x00\x03\x00keyval'),
ffs(br'1 command-argument eoa \x03\x00\x03\x00foobar'), ffs(br'1 1 0 command-argument eoa \x03\x00\x03\x00foobar'),
ffs(b'1 command-data continuation value1'), ffs(b'1 1 0 command-data continuation value1'),
ffs(b'1 command-data eos value2'), ffs(b'1 1 0 command-data eos value2'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertaction(results[-1], 'runcommand') self.assertaction(results[-1], 'runcommand')
self.assertEqual(results[-1][1], { self.assertEqual(results[-1][1], {
'requestid': 1, 'requestid': 1,
'command': b'command', 'command': b'command',
'args': { 'args': {
b'key': b'val', b'key': b'val',
b'foo': b'bar', b'foo': b'bar',
}, },
'data': b'value1value2', 'data': b'value1value2',
}) })
def testunexpectedcommandargument(self): def testunexpectedcommandargument(self):
"""Command argument frame when not running a command is an error.""" """Command argument frame when not running a command is an error."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(
ffs(b'1 command-argument 0 ignored')) makereactor(), ffs(b'1 1 stream-begin command-argument 0 ignored'))
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'expected command frame; got 2', 'message': b'expected command frame; got 2',
}) })
def testunexpectedcommandargumentreceiving(self): def testunexpectedcommandargumentreceiving(self):
"""Same as above but the command is receiving.""" """Same as above but the command is receiving."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-data command'), ffs(b'1 1 stream-begin command-name have-data command'),
ffs(b'1 command-argument eoa ignored'), ffs(b'1 1 0 command-argument eoa ignored'),
])) ]))
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'received command argument frame for request that is ' 'message': b'received command argument frame for request that is '
b'not expecting arguments: 1', b'not expecting arguments: 1',
}) })
def testunexpectedcommanddata(self): def testunexpectedcommanddata(self):
"""Command argument frame when not running a command is an error.""" """Command argument frame when not running a command is an error."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(
ffs(b'1 command-data 0 ignored')) makereactor(), ffs(b'1 1 stream-begin command-data 0 ignored'))
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'expected command frame; got 3', 'message': b'expected command frame; got 3',
}) })
def testunexpectedcommanddatareceiving(self): def testunexpectedcommanddatareceiving(self):
"""Same as above except the command is receiving.""" """Same as above except the command is receiving."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command'), ffs(b'1 1 stream-begin command-name have-args command'),
ffs(b'1 command-data eos ignored'), ffs(b'1 1 0 command-data eos ignored'),
])) ]))
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'received command data frame for request that is not ' 'message': b'received command data frame for request that is not '
b'expecting data: 1', b'expecting data: 1',
}) })
def testmissingcommandframeflags(self): def testmissingcommandframeflags(self):
"""Command name frame must have flags set.""" """Command name frame must have flags set."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(
ffs(b'1 command-name 0 command')) makereactor(), ffs(b'1 1 stream-begin command-name 0 command'))
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'missing frame flags on command frame', 'message': b'missing frame flags on command frame',
}) })
def testconflictingrequestidallowed(self): def testconflictingrequestidallowed(self):
"""Multiple fully serviced commands with same request ID is allowed.""" """Multiple fully serviced commands with same request ID is allowed."""
reactor = makereactor() reactor = makereactor()
results = [] results = []
outstream = framing.stream() outstream = framing.stream(2)
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 1 stream-begin command-name eos command')))
result = reactor.onbytesresponseready(outstream, 1, b'response1') result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 1 0 command-name eos command')))
result = reactor.onbytesresponseready(outstream, 1, b'response2') result = reactor.onbytesresponseready(outstream, 1, b'response2')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 command-name eos command'))) reactor, ffs(b'1 1 0 command-name eos command')))
result = reactor.onbytesresponseready(outstream, 1, b'response3') result = reactor.onbytesresponseready(outstream, 1, b'response3')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
for i in range(3): for i in range(3):
self.assertaction(results[i], 'runcommand') self.assertaction(results[i], 'runcommand')
self.assertEqual(results[i][1], { self.assertEqual(results[i][1], {
'requestid': 1, 'requestid': 1,
'command': b'command', 'command': b'command',
'args': {}, 'args': {},
'data': None, 'data': None,
}) })
def testconflictingrequestid(self): def testconflictingrequestid(self):
"""Request ID for new command matching in-flight command is illegal.""" """Request ID for new command matching in-flight command is illegal."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command'), ffs(b'1 1 stream-begin command-name have-args command'),
ffs(b'1 command-name eos command'), ffs(b'1 1 0 command-name eos command'),
])) ]))
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'request with ID 1 already received', 'message': b'request with ID 1 already received',
}) })
def testinterleavedcommands(self): def testinterleavedcommands(self):
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command1'), ffs(b'1 1 stream-begin command-name have-args command1'),
ffs(b'3 command-name have-args command3'), ffs(b'3 1 0 command-name have-args command3'),
ffs(br'1 command-argument 0 \x03\x00\x03\x00foobar'), ffs(br'1 1 0 command-argument 0 \x03\x00\x03\x00foobar'),
ffs(br'3 command-argument 0 \x03\x00\x03\x00bizbaz'), ffs(br'3 1 0 command-argument 0 \x03\x00\x03\x00bizbaz'),
ffs(br'3 command-argument eoa \x03\x00\x03\x00keyval'), ffs(br'3 1 0 command-argument eoa \x03\x00\x03\x00keyval'),
ffs(br'1 command-argument eoa \x04\x00\x03\x00key1val'), ffs(br'1 1 0 command-argument eoa \x04\x00\x03\x00key1val'),
])) ]))
self.assertEqual([t[0] for t in results], [ self.assertEqual([t[0] for t in results], [
'wantframe', 'wantframe',
'wantframe', 'wantframe',
'wantframe', 'wantframe',
'wantframe', 'wantframe',
'runcommand', 'runcommand',
'data': None, 'data': None,
}) })
def testmissingargumentframe(self): def testmissingargumentframe(self):
# This test attempts to test behavior when reactor has an incomplete # This test attempts to test behavior when reactor has an incomplete
# command request waiting on argument data. But it doesn't handle that # command request waiting on argument data. But it doesn't handle that
# scenario yet. So this test does nothing of value. # scenario yet. So this test does nothing of value.
frames = [ frames = [
ffs(b'1 command-name have-args command'), ffs(b'1 1 stream-begin command-name have-args command'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
def testincompleteargumentname(self): def testincompleteargumentname(self):
"""Argument frame with incomplete name.""" """Argument frame with incomplete name."""
frames = [ frames = [
ffs(b'1 command-name have-args command1'), ffs(b'1 1 stream-begin command-name have-args command1'),
ffs(br'1 command-argument eoa \x04\x00\xde\xadfoo'), ffs(br'1 1 0 command-argument eoa \x04\x00\xde\xadfoo'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'malformed argument frame: partial argument name', 'message': b'malformed argument frame: partial argument name',
}) })
def testincompleteargumentvalue(self): def testincompleteargumentvalue(self):
"""Argument frame with incomplete value.""" """Argument frame with incomplete value."""
frames = [ frames = [
ffs(b'1 command-name have-args command'), ffs(b'1 1 stream-begin command-name have-args command'),
ffs(br'1 command-argument eoa \x03\x00\xaa\xaafoopartialvalue'), ffs(br'1 1 0 command-argument eoa \x03\x00\xaa\xaafoopartialvalue'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'malformed argument frame: partial argument value', 'message': b'malformed argument frame: partial argument value',
}) })
def testmissingcommanddataframe(self): def testmissingcommanddataframe(self):
# The reactor doesn't currently handle partially received commands. # The reactor doesn't currently handle partially received commands.
# So this test is failing to do anything with request 1. # So this test is failing to do anything with request 1.
frames = [ frames = [
ffs(b'1 command-name have-data command1'), ffs(b'1 1 stream-begin command-name have-data command1'),
ffs(b'3 command-name eos command2'), ffs(b'3 1 0 command-name eos command2'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
def testmissingcommanddataframeflags(self): def testmissingcommanddataframeflags(self):
frames = [ frames = [
ffs(b'1 command-name have-data command1'), ffs(b'1 1 stream-begin command-name have-data command1'),
ffs(b'1 command-data 0 data'), ffs(b'1 1 0 command-data 0 data'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'command data frame without flags', 'message': b'command data frame without flags',
}) })
def testframefornonreceivingrequest(self): def testframefornonreceivingrequest(self):
"""Receiving a frame for a command that is not receiving is illegal.""" """Receiving a frame for a command that is not receiving is illegal."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 command-name eos command1'), ffs(b'1 1 stream-begin command-name eos command1'),
ffs(b'3 command-name have-data command3'), ffs(b'3 1 0 command-name have-data command3'),
ffs(b'5 command-argument eoa ignored'), ffs(b'5 1 0 command-argument eoa ignored'),
])) ]))
self.assertaction(results[2], 'error') self.assertaction(results[2], 'error')
self.assertEqual(results[2][1], { self.assertEqual(results[2][1], {
'message': b'received frame for request that is not receiving: 5', 'message': b'received frame for request that is not receiving: 5',
}) })
def testsimpleresponse(self): def testsimpleresponse(self):
"""Bytes response to command sends result frames.""" """Bytes response to command sends result frames."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream() outstream = framing.stream(2)
result = reactor.onbytesresponseready(outstream, 1, b'response') result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response', b'1 2 stream-begin bytes-response eos response',
]) ])
def testmultiframeresponse(self): def testmultiframeresponse(self):
"""Bytes response spanning multiple frames is handled.""" """Bytes response spanning multiple frames is handled."""
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE
second = b'y' * 100 second = b'y' * 100
reactor = makereactor() reactor = makereactor()
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream() outstream = framing.stream(2)
result = reactor.onbytesresponseready(outstream, 1, first + second) result = reactor.onbytesresponseready(outstream, 1, first + second)
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response continuation %s' % first, b'1 2 stream-begin bytes-response continuation %s' % first,
b'1 bytes-response eos %s' % second, b'1 2 0 bytes-response eos %s' % second,
]) ])
def testapplicationerror(self): def testapplicationerror(self):
reactor = makereactor() reactor = makereactor()
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream() outstream = framing.stream(2)
result = reactor.onapplicationerror(outstream, 1, b'some message') result = reactor.onapplicationerror(outstream, 1, b'some message')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 error-response application some message', b'1 2 stream-begin error-response application some message',
]) ])
def test1commanddeferresponse(self): def test1commanddeferresponse(self):
"""Responses when in deferred output mode are delayed until EOF.""" """Responses when in deferred output mode are delayed until EOF."""
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream() instream = framing.stream(1)
results = list(sendcommandframes(reactor, instream, 1, b'mycommand', results = list(sendcommandframes(reactor, instream, 1, b'mycommand',
{})) {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
outstream = framing.stream() outstream = framing.stream(2)
result = reactor.onbytesresponseready(outstream, 1, b'response') result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response', b'1 2 stream-begin bytes-response eos response',
]) ])
def testmultiplecommanddeferresponse(self): def testmultiplecommanddeferresponse(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, instream, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 3, b'command2', {}))
outstream = framing.stream() outstream = framing.stream(2)
result = reactor.onbytesresponseready(outstream, 1, b'response1') result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.onbytesresponseready(outstream, 3, b'response2') result = reactor.onbytesresponseready(outstream, 3, b'response2')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 bytes-response eos response1', b'1 2 stream-begin bytes-response eos response1',
b'3 bytes-response eos response2' b'3 2 0 bytes-response eos response2'
]) ])
def testrequestidtracking(self): def testrequestidtracking(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, instream, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 3, b'command2', {}))
list(sendcommandframes(reactor, instream, 5, b'command3', {})) list(sendcommandframes(reactor, instream, 5, b'command3', {}))
# Register results for commands out of order. # Register results for commands out of order.
outstream = framing.stream() outstream = framing.stream(2)
reactor.onbytesresponseready(outstream, 3, b'response3') reactor.onbytesresponseready(outstream, 3, b'response3')
reactor.onbytesresponseready(outstream, 1, b'response1') reactor.onbytesresponseready(outstream, 1, b'response1')
reactor.onbytesresponseready(outstream, 5, b'response5') reactor.onbytesresponseready(outstream, 5, b'response5')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'3 bytes-response eos response3', b'3 2 stream-begin bytes-response eos response3',
b'1 bytes-response eos response1', b'1 2 0 bytes-response eos response1',
b'5 bytes-response eos response5', b'5 2 0 bytes-response eos response5',
]) ])
def testduplicaterequestonactivecommand(self): def testduplicaterequestonactivecommand(self):
"""Receiving a request ID that matches a request that isn't finished.""" """Receiving a request ID that matches a request that isn't finished."""
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
list(sendcommandframes(reactor, stream, 1, b'command1', {})) list(sendcommandframes(reactor, stream, 1, b'command1', {}))
results = list(sendcommandframes(reactor, stream, 1, b'command1', {})) results = list(sendcommandframes(reactor, stream, 1, b'command1', {}))
self.assertaction(results[0], 'error') self.assertaction(results[0], 'error')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'message': b'request with ID 1 is already active', 'message': b'request with ID 1 is already active',
}) })
def testduplicaterequestonactivecommandnosend(self): def testduplicaterequestonactivecommandnosend(self):
"""Same as above but we've registered a response but haven't sent it.""" """Same as above but we've registered a response but haven't sent it."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream() outstream = framing.stream(2)
reactor.onbytesresponseready(outstream, 1, b'response') reactor.onbytesresponseready(outstream, 1, b'response')
# We've registered the response but haven't sent it. From the # We've registered the response but haven't sent it. From the
# perspective of the reactor, the command is still active. # perspective of the reactor, the command is still active.
results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'error') self.assertaction(results[0], 'error')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'message': b'request with ID 1 is already active', 'message': b'request with ID 1 is already active',
}) })
def testduplicaterequestargumentframe(self): def testduplicaterequestargumentframe(self):
"""Variant on above except we sent an argument frame instead of name.""" """Variant on above except we sent an argument frame instead of name."""
reactor = makereactor() reactor = makereactor()
stream = framing.stream() stream = framing.stream(1)
list(sendcommandframes(reactor, stream, 1, b'command', {})) list(sendcommandframes(reactor, stream, 1, b'command', {}))
results = list(sendframes(reactor, [ results = list(sendframes(reactor, [
ffs(b'3 command-name have-args command'), ffs(b'3 1 stream-begin command-name have-args command'),
ffs(b'1 command-argument 0 ignored'), ffs(b'1 1 0 command-argument 0 ignored'),
])) ]))
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': 'received frame for request that is still active: 1', 'message': 'received frame for request that is still active: 1',
}) })
def testduplicaterequestaftersend(self): def testduplicaterequestaftersend(self):
"""We can use a duplicate request ID after we've sent the response.""" """We can use a duplicate request ID after we've sent the response."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream() instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream() outstream = framing.stream(2)
res = reactor.onbytesresponseready(outstream, 1, b'response') res = reactor.onbytesresponseready(outstream, 1, b'response')
list(res[1]['framegen']) list(res[1]['framegen'])
results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)