This is an archive of the discontinued Mercurial Phabricator instance.

Status	Author	Revision
Closed	indygreg	D3243 httppeer: support protocol upgrade
Closed	indygreg	D3242 wireproto: define and implement HTTP handshake to upgrade protocol
Closed	indygreg	D3241 httppeer: only advertise partial-pull if capabilities are known
Closed	indygreg	D3240 httppeer: always add x-hg* headers to Vary header
Closed	indygreg	D3239 httppeer: don't accept very old media types (BC)
Closed	indygreg	D3238 httppeer: allow opener to be passed to makepeer()
Closed	indygreg	D3237 httppeer: perform capabilities request in makepeer()
Closed	indygreg	D3236 httppeer: extract common response handling into own function
Closed	indygreg	D3235 httppeer: move error handling and response wrapping into sendrequest
Closed	indygreg	D3234 httppeer: extract code for creating a request into own function
Closed	indygreg	D3233 httppeer: extract code for performing an HTTP request
Closed	indygreg	D3232 httppeer: move requestbuilder defaults into makepeer() argument
Closed	indygreg	D3231 wireproto: move version 2 command handlers to wireprotov2server
Closed	indygreg	D3230 wireproto: extract HTTP version 2 code to own module
Closed	indygreg	D3224 wireproto: client reactor support for receiving frames
Closed	indygreg	D3223 wireproto: introduce a reactor for client-side state
Closed	indygreg	D3222 tests: extract wire protocol framing tests to own file
Closed	indygreg	D3208 wireproto: disallow commands handlers for multiple transport versions
Closed	indygreg	D3207 wireproto: make @wireprotocommand version 1 only by default
Closed	indygreg	D3206 wireproto: only expose "getbundle" and "unbundle" to v1 transports
Closed	indygreg	D3205 wireproto: port lookup to wire protocol v2
Closed	indygreg	D3204 wireproto: port pushkey command to wire protocol version 2
Closed	indygreg	D3203 wireproto: only expose "clonebundles" to version 1 transports
Closed	indygreg	D3202 wireproto: define and expose types of wire command arguments
Closed	indygreg	D3201 wireproto: only expose "stream_out" to version 1 transports
Closed	indygreg	D3200 wireproto: implement capabilities for wire protocol v2

Diff 7942

mercurial/httppeer.py


	while True:			while True:
	frame = wireprotoframing.readframe(res)			frame = wireprotoframing.readframe(res)
	if frame is None:			if frame is None:
	break			break

	self.ui.note(_('received %r\n') % frame)			self.ui.note(_('received %r\n') % frame)

	if frame.typeid == wireprotoframing.FRAME_TYPE_BYTES_RESPONSE:			action, meta = reactor.onframerecv(frame)
	if frame.flags & wireprotoframing.FLAG_BYTES_RESPONSE_CBOR:
	payload = util.bytesio(frame.payload)			if action == 'responsedata':
				if meta['cbor']:
				payload = util.bytesio(meta['data'])

	decoder = cbor.CBORDecoder(payload)			decoder = cbor.CBORDecoder(payload)
	while payload.tell() + 1 < len(frame.payload):			while payload.tell() + 1 < len(meta['data']):
	results.append(decoder.decode())			results.append(decoder.decode())
	else:			else:
	results.append(frame.payload)			results.append(meta['data'])
	else:			else:
	error.ProgrammingError('unhandled frame type: %d' %			error.ProgrammingError('unhandled action: %s' % action)
	frame.typeid)

	return results			return results

	def makepeer(ui, path):			def makepeer(ui, path):
	u = util.url(path)			u = util.url(path)
	if u.query or u.fragment:			if u.query or u.fragment:
	raise error.Abort(_('unsupported URL component: "%s"') %			raise error.Abort(_('unsupported URL component: "%s"') %
	(u.query or u.fragment))			(u.query or u.fragment))

mercurial/wireprotoframing.py

	self._canissuecommands = True			self._canissuecommands = True
	self._cansend = True			self._cansend = True

	self._nextrequestid = 1			self._nextrequestid = 1
	# We only support a single outgoing stream for now.			# We only support a single outgoing stream for now.
	self._outgoingstream = stream(1)			self._outgoingstream = stream(1)
	self._pendingrequests = collections.deque()			self._pendingrequests = collections.deque()
	self._activerequests = {}			self._activerequests = {}
				self._incomingstreams = {}

	def callcommand(self, name, args, datafh=None):			def callcommand(self, name, args, datafh=None):
	"""Request that a command be executed.			"""Request that a command be executed.

	Receives the command name, a dict of arguments to pass to the command,			Receives the command name, a dict of arguments to pass to the command,
	and an optional file object containing the raw data for the command.			and an optional file object containing the raw data for the command.

	Returns a 3-tuple of (request, action, action data).			Returns a 3-tuple of (request, action, action data).
	request.name,			request.name,
	request.args,			request.args,
	request.datafh)			request.datafh)

	for frame in res:			for frame in res:
	yield frame			yield frame

	request.state = 'sent'			request.state = 'sent'

				def onframerecv(self, frame):
				"""Process a frame that has been received off the wire.

				Returns a 2-tuple of (action, meta) describing further action the
				caller needs to take as a result of receiving this frame.
				"""
				if frame.streamid % 2:
				return 'error', {
				'message': (
				_('received frame with odd numbered stream ID: %d') %
				frame.streamid),
				}

				if frame.streamid not in self._incomingstreams:
				if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
				return 'error', {
				'message': _('received frame on unknown stream '
				'without beginning of stream flag set'),
				}

				if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
				raise error.ProgrammingError('support for decoding stream '
				'payloads not yet implemneted')

				if frame.streamflags & STREAM_FLAG_END_STREAM:
				del self._incomingstreams[frame.streamid]

				if frame.requestid not in self._activerequests:
				return 'error', {
				'message': (_('received frame for inactive request ID: %d') %
				frame.requestid),
				}

				request = self._activerequests[frame.requestid]
				request.state = 'receiving'

				handlers = {
				FRAME_TYPE_BYTES_RESPONSE: self._onbytesresponseframe,
				}

				meth = handlers.get(frame.typeid)
				if not meth:
				raise error.ProgrammingError('unhandled frame type: %d' %
				frame.typeid)

				return meth(request, frame)

				def _onbytesresponseframe(self, request, frame):
				if frame.flags & FLAG_BYTES_RESPONSE_EOS:
				request.state = 'received'
				del self._activerequests[request.requestid]

				return 'responsedata', {
				'request': request,
				'expectmore': frame.flags & FLAG_BYTES_RESPONSE_CONTINUATION,
				'eos': frame.flags & FLAG_BYTES_RESPONSE_EOS,
				'cbor': frame.flags & FLAG_BYTES_RESPONSE_CBOR,
				'data': frame.payload,
				}

tests/test-wireproto-clientreactor.py

	from __future__ import absolute_import			from __future__ import absolute_import

	import unittest			import unittest

	from mercurial import (			from mercurial import (
	error,			error,
	wireprotoframing as framing,			wireprotoframing as framing,
	)			)

				ffs = framing.makeframefromhumanstring

				def sendframe(reactor, frame):
				"""Send a frame bytearray to a reactor."""
				header = framing.parseheader(frame)
				payload = frame[framing.FRAME_HEADER_SIZE:]
				assert len(payload) == header.length

				return reactor.onframerecv(framing.frame(header.requestid,
				header.streamid,
				header.streamflags,
				header.typeid,
				header.flags,
				payload))

	class SingleSendTests(unittest.TestCase):			class SingleSendTests(unittest.TestCase):
	"""A reactor that can only send once rejects subsequent sends."""			"""A reactor that can only send once rejects subsequent sends."""
	def testbasic(self):			def testbasic(self):
	reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True)			reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True)

	request, action, meta = reactor.callcommand(b'foo', {})			request, action, meta = reactor.callcommand(b'foo', {})
	self.assertEqual(request.state, 'pending')			self.assertEqual(request.state, 'pending')
	self.assertEqual(action, 'noop')			self.assertEqual(action, 'noop')
	self.assertEqual(request.requestid, 3)			self.assertEqual(request.requestid, 3)
	self.assertEqual(action, 'sendframes')			self.assertEqual(action, 'sendframes')

	for frame in meta['framegen']:			for frame in meta['framegen']:
	self.assertEqual(request.state, 'sending')			self.assertEqual(request.state, 'sending')

	self.assertEqual(request.state, 'sent')			self.assertEqual(request.state, 'sent')

				class BadFrameRecvTests(unittest.TestCase):
				def testoddstream(self):
				reactor = framing.clientreactor()

				action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo'))
				self.assertEqual(action, 'error')
				self.assertEqual(meta['message'],
				'received frame with odd numbered stream ID: 1')

				def testunknownstream(self):
				reactor = framing.clientreactor()

				action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo'))
				self.assertEqual(action, 'error')
				self.assertEqual(meta['message'],
				'received frame on unknown stream without beginning '
				'of stream flag set')

				def testunhandledframetype(self):
				reactor = framing.clientreactor(buffersends=False)

				request, action, meta = reactor.callcommand(b'foo', {})
				for frame in meta['framegen']:
				pass

				with self.assertRaisesRegexp(error.ProgrammingError,
				'unhandled frame type'):
				sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo'))

	if __name__ == '__main__':			if __name__ == '__main__':
	import silenttestrunner			import silenttestrunner
	silenttestrunner.main(__name__)			silenttestrunner.main(__name__)

Diff	ID	Description	Created	Lint	Unit
Base		Base
Diff 1	7942		Apr 10 2018, 9:26 PM	★	★
Diff 2	7990	rHG55b5ba8d4e6875a6f81052020f4be5d397a6888f	Apr 9 2018, 7:54 PM	★	★

This is an archive of the discontinued Mercurial Phabricator instance.

wireproto: client reactor support for receiving framesClosedPublic

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 7942

mercurial/httppeer.py

mercurial/wireprotoframing.py

tests/test-wireproto-clientreactor.py

wireproto: client reactor support for receiving frames
ClosedPublic

Revision Contents
Changeset List