| hg-reviewers |
I think this makes the code a bit clearer. And documentation is
always useful, right?
| Lint Skipped |
| Unit Tests Skipped |
This one failed to apply. It kind of looks like it got folded into earlier revisions, is that right?
Yes. I folded this in to previous commits and hg phabsend wasn't smart enough to realize it.
| reactionary to external events. The point of this class is to encapsulate | reactionary to external events. The point of this class is to encapsulate | ||||
| the state of the connection and the exchange of frames, not to perform | the state of the connection and the exchange of frames, not to perform | ||||
| work. Instead, callers tell this class when something occurs, like a | work. Instead, callers tell this class when something occurs, like a | ||||
| frame arriving. If that activity is worthy of a follow-up action (say | frame arriving. If that activity is worthy of a follow-up action (say | ||||
| *run a command*), the return value of that handler will say so. | *run a command*), the return value of that handler will say so. | ||||
| I/O and CPU intensive operations are purposefully delegated outside of | I/O and CPU intensive operations are purposefully delegated outside of | ||||
| this class. | this class. | ||||
| Consumers are expected to tell instances when events occur. They do so by | |||||
| calling the various ``on*`` methods. These methods return a 2-tuple | |||||
| describing any follow-up action(s) to take. The first element is the | |||||
| name of an action to perform. The second is a data structure (usually | |||||
| a dict) specific to that action that contains more information. e.g. | |||||
| if the server wants to send frames back to the client, the data structure | |||||
| will contain a reference to those frames. | |||||
| Valid actions that consumers can be instructed to take are: | |||||
| sendframes | |||||
| Indicates that frames should be sent to the client. The ``framegen`` | |||||
| key contains a generator of frames that should be sent. The server | |||||
| assumes that all frames are sent to the client. | |||||
| error | |||||
| Indicates that an error occurred. Consumer should probably abort. | |||||
| runcommand | |||||
| Indicates that the consumer should run a wire protocol command. Details | |||||
| of the command to run are given in the data structure. | |||||
| wantframe | |||||
| Indicates that nothing of interest happened and the server is waiting on | |||||
| more frames from the client before anything interesting can be done. | |||||
| """ | """ | ||||
| def __init__(self, ui, repo): | def __init__(self, ui, repo): | ||||
| self._ui = ui | self._ui = ui | ||||
| self._repo = repo | self._repo = repo | ||||
| self._state = 'idle' | self._state = 'idle' | ||||
| self._activecommand = None | self._activecommand = None | ||||
| self._activeargs = None | self._activeargs = None | ||||
| return meth(frametype, frameflags, payload) | return meth(frametype, frameflags, payload) | ||||
| def onbytesresponseready(self, data): | def onbytesresponseready(self, 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. | ||||
| """ | """ | ||||
| return { | return 'sendframes', { | ||||
| 'action': 'sendframes', | |||||
| 'framegen': createbytesresponseframesfrombytes(data), | 'framegen': createbytesresponseframesfrombytes(data), | ||||
| } | } | ||||
| def onapplicationerror(self, msg): | def onapplicationerror(self, msg): | ||||
| return { | return 'sendframes', { | ||||
| 'action': 'sendframes', | |||||
| 'framegen': createerrorframe(msg, application=True), | 'framegen': createerrorframe(msg, application=True), | ||||
| } | } | ||||
| def _makeerrorresult(self, msg): | def _makeerrorresult(self, msg): | ||||
| return { | return 'error', { | ||||
| 'action': 'error', | |||||
| 'message': msg, | 'message': msg, | ||||
| } | } | ||||
| def _makeruncommandresult(self): | def _makeruncommandresult(self): | ||||
| return { | return 'runcommand', { | ||||
| 'action': 'runcommand', | |||||
| 'command': self._activecommand, | 'command': self._activecommand, | ||||
| 'args': self._activeargs, | 'args': self._activeargs, | ||||
| 'data': self._activedata.getvalue() if self._activedata else None, | 'data': self._activedata.getvalue() if self._activedata else None, | ||||
| } | } | ||||
| def _makewantframeresult(self): | def _makewantframeresult(self): | ||||
| return { | return 'wantframe', { | ||||
| 'action': 'wantframe', | 'action': 'wantframe', | ||||
| 'state': self._state, | 'state': self._state, | ||||
| } | } | ||||
| def _onframeidle(self, frametype, frameflags, payload): | def _onframeidle(self, frametype, frameflags, payload): | ||||
| # The only frame type that should be received in this state is a | # The only frame type that should be received in this state is a | ||||
| # command request. | # command request. | ||||
| if frametype != FRAME_TYPE_COMMAND_NAME: | if frametype != FRAME_TYPE_COMMAND_NAME: | ||||
| if not frame: | if not frame: | ||||
| states.append(b'received: <no frame>') | states.append(b'received: <no frame>') | ||||
| break | break | ||||
| frametype, frameflags, payload = frame | frametype, frameflags, payload = frame | ||||
| states.append(b'received: %d %d %s' % (frametype, frameflags, payload)) | states.append(b'received: %d %d %s' % (frametype, frameflags, payload)) | ||||
| r = reactor.onframerecv(frametype, frameflags, payload) | action, meta = reactor.onframerecv(frametype, frameflags, payload) | ||||
| states.append(json.dumps(r, sort_keys=True, separators=(', ', ': '))) | meta['action'] = action | ||||
| states.append(json.dumps(meta, sort_keys=True, separators=(', ', ': '))) | |||||
| res.status = b'200 OK' | res.status = b'200 OK' | ||||
| res.headers[b'Content-Type'] = b'text/plain' | res.headers[b'Content-Type'] = b'text/plain' | ||||
| res.setbodybytes(b'\n'.join(states)) | res.setbodybytes(b'\n'.join(states)) | ||||
| def _processhttpv2request(ui, repo, req, res, authedperm, reqcommand, proto): | def _processhttpv2request(ui, repo, req, res, authedperm, reqcommand, proto): | ||||
| """Post-validation handler for HTTPv2 requests. | """Post-validation handler for HTTPv2 requests. | ||||
| Called when the HTTP request contains unified frame-based protocol | Called when the HTTP request contains unified frame-based protocol | ||||
| frames for evaluation. | frames for evaluation. | ||||
| """ | """ | ||||
| reactor = wireprotoframing.serverreactor(ui, repo) | reactor = wireprotoframing.serverreactor(ui, repo) | ||||
| seencommand = False | seencommand = False | ||||
| while True: | while True: | ||||
| frame = wireprotoframing.readframe(req.bodyfh) | frame = wireprotoframing.readframe(req.bodyfh) | ||||
| if not frame: | if not frame: | ||||
| break | break | ||||
| state = reactor.onframerecv(*frame) | action, meta = reactor.onframerecv(*frame) | ||||
| if not state: | |||||
| raise error.ProgrammingError( | |||||
| 'did not receive valid reply from server reactor') | |||||
| action = state['action'] | |||||
| if action == 'wantframe': | if action == 'wantframe': | ||||
| # Need more data before we can do anything. | # Need more data before we can do anything. | ||||
| continue | continue | ||||
| elif action == 'runcommand': | elif action == 'runcommand': | ||||
| # We currently only support running a single command per | # We currently only support running a single command per | ||||
| # HTTP request. | # HTTP request. | ||||
| if seencommand: | if seencommand: | ||||
| # TODO define proper error mechanism. | # TODO define proper error mechanism. | ||||
| res.status = b'200 OK' | res.status = b'200 OK' | ||||
| res.headers[b'Content-Type'] = b'text/plain' | res.headers[b'Content-Type'] = b'text/plain' | ||||
| res.setbodybytes(_('support for multiple commands per request ' | res.setbodybytes(_('support for multiple commands per request ' | ||||
| 'not yet implemented')) | 'not yet implemented')) | ||||
| return | return | ||||
| _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, | _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, | ||||
| reactor, state) | reactor, meta) | ||||
| elif action == 'error': | elif action == 'error': | ||||
| # TODO define proper error mechanism. | # TODO define proper error mechanism. | ||||
| res.status = b'200 OK' | res.status = b'200 OK' | ||||
| res.headers[b'Content-Type'] = b'text/plain' | res.headers[b'Content-Type'] = b'text/plain' | ||||
| res.setbodybytes(state['message'] + b'\n') | res.setbodybytes(meta['message'] + b'\n') | ||||
| return | return | ||||
| else: | else: | ||||
| raise error.ProgrammingError( | raise error.ProgrammingError( | ||||
| 'unhandled action from frame processor: %s' % action) | 'unhandled action from frame processor: %s' % action) | ||||
| def _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, reactor, | def _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, reactor, | ||||
| command): | command): | ||||
| """Dispatch a wire protocol command made from HTTPv2 requests. | """Dispatch a wire protocol command made from HTTPv2 requests. | ||||
| assert wirecommand.permission in ('push', 'pull') | assert wirecommand.permission in ('push', 'pull') | ||||
| 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 | ||||
| if isinstance(rsp, wireprototypes.bytesresponse): | if isinstance(rsp, wireprototypes.bytesresponse): | ||||
| action = reactor.onbytesresponseready(rsp.data) | action, meta = reactor.onbytesresponseready(rsp.data) | ||||
| else: | else: | ||||
| action = reactor.onapplicationerror( | action, meta = reactor.onapplicationerror( | ||||
| _('unhandled response type from wire proto command')) | _('unhandled response type from wire proto command')) | ||||
| if action['action'] == b'sendframes': | if action == 'sendframes': | ||||
| res.setbodygen(action['framegen']) | res.setbodygen(meta['framegen']) | ||||
| else: | else: | ||||
| raise error.ProgrammingError('unhandled event from reactor: %s' % | raise error.ProgrammingError('unhandled event from reactor: %s' % | ||||
| action['action']) | action) | ||||
| # Maps API name to metadata so custom API can be registered. | # Maps API name to metadata so custom API can be registered. | ||||
| API_HANDLERS = { | API_HANDLERS = { | ||||
| HTTPV2: { | HTTPV2: { | ||||
| 'config': ('experimental', 'web.api.http-v2'), | 'config': ('experimental', 'web.api.http-v2'), | ||||
| 'handler': _handlehttpv2request, | 'handler': _handlehttpv2request, | ||||
| }, | }, | ||||
| } | } | ||||