| hg-reviewers |
In the near future, we will want to support performing I/O from
other sources. Let's rename readframe() to readdata() and tweak
its logic to support future growth.
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| f.set_exception(error.ResponseError( | f.set_exception(error.ResponseError( | ||||
| _('unfulfilled command response'))) | _('unfulfilled command response'))) | ||||
| self._futures = None | self._futures = None | ||||
| def _handleresponse(self, handler, resp): | def _handleresponse(self, handler, resp): | ||||
| # Called in a thread to read the response. | # Called in a thread to read the response. | ||||
| while handler.readframe(resp): | while handler.readdata(resp): | ||||
| pass | pass | ||||
| # TODO implement interface for version 2 peers | # TODO implement interface for version 2 peers | ||||
| @interfaceutil.implementer(repository.ipeerconnection, | @interfaceutil.implementer(repository.ipeerconnection, | ||||
| repository.ipeercapabilities, | repository.ipeercapabilities, | ||||
| repository.ipeerrequests) | repository.ipeerrequests) | ||||
| class httpv2peer(object): | class httpv2peer(object): | ||||
| def __init__(self, ui, repourl, apipath, opener, requestbuilder, | def __init__(self, ui, repourl, apipath, opener, requestbuilder, | ||||
| """ | """ | ||||
| def __init__(self, ui, clientreactor): | def __init__(self, ui, clientreactor): | ||||
| self._ui = ui | self._ui = ui | ||||
| self._reactor = clientreactor | self._reactor = clientreactor | ||||
| self._requests = {} | self._requests = {} | ||||
| self._futures = {} | self._futures = {} | ||||
| self._responses = {} | self._responses = {} | ||||
| self._frameseof = False | |||||
| def callcommand(self, command, args, f): | def callcommand(self, command, args, f): | ||||
| """Register a request to call a command. | """Register a request to call a command. | ||||
| Returns an iterable of frames that should be sent over the wire. | Returns an iterable of frames that should be sent over the wire. | ||||
| """ | """ | ||||
| request, action, meta = self._reactor.callcommand(command, args) | request, action, meta = self._reactor.callcommand(command, args) | ||||
| """ | """ | ||||
| action, meta = self._reactor.flushcommands() | action, meta = self._reactor.flushcommands() | ||||
| if action != 'sendframes': | if action != 'sendframes': | ||||
| raise error.ProgrammingError('%s not yet supported' % action) | raise error.ProgrammingError('%s not yet supported' % action) | ||||
| return meta['framegen'] | return meta['framegen'] | ||||
| def readframe(self, fh): | def readdata(self, framefh): | ||||
| """Attempt to read and process a frame. | """Attempt to read data and do work. | ||||
| Returns None if no frame was read. Presumably this means EOF. | Returns None if no data was read. Presumably this means we're | ||||
| done with all read I/O. | |||||
| """ | """ | ||||
| frame = wireprotoframing.readframe(fh) | if not self._frameseof: | ||||
| frame = wireprotoframing.readframe(framefh) | |||||
| if frame is None: | if frame is None: | ||||
| # TODO tell reactor? | # TODO tell reactor? | ||||
| return | self._frameseof = True | ||||
| else: | |||||
| self._ui.note(_('received %r\n') % frame) | self._ui.note(_('received %r\n') % frame) | ||||
| self._processframe(frame) | self._processframe(frame) | ||||
| if self._frameseof: | |||||
| return None | |||||
| return True | return True | ||||
| def _processframe(self, frame): | def _processframe(self, frame): | ||||
| """Process a single read frame.""" | """Process a single read frame.""" | ||||
| action, meta = self._reactor.onframerecv(frame) | action, meta = self._reactor.onframerecv(frame) | ||||
| if action == 'error': | if action == 'error': | ||||