mercurial/httppeer.py

37718:ad1c07008e0b

import errno
import io
import os
import socket
import struct
import sys
import tempfile
import weakref

from .i18n import _
from .thirdparty import (

    url as urlmod,
    util,
    wireprotoframing,
    wireprototypes,
    wireprotov1peer,
    wireprotov2server,
)

httplib = util.httplib
urlerr = util.urlerr

def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests):
    reactor = wireprotoframing.clientreactor(hasmultiplesend=False,
                                             buffersends=True)

    url = '%s/%s' % (apiurl, permission)

    if len(requests) > 1:
        url += '/multirequest'
    else:
        url += '/%s' % requests[0][0]

    # Request ID to (request, future)
    requestmap = {}

    for command, args, f in requests:
        request, action, meta = reactor.callcommand(command, args)
        assert action == 'noop'

        requestmap[request.requestid] = (request, f)

    action, meta = reactor.flushcommands()
    assert action == 'sendframes'

    # TODO stream this.
    body = b''.join(map(bytes, meta['framegen']))

    # TODO modify user-agent to reflect v2
    headers = {
        r'Accept': wireprotov2server.FRAMINGTYPE,
        r'Content-Type': wireprotov2server.FRAMINGTYPE,

        raise
    except httplib.HTTPException as e:
        ui.traceback()
        raise IOError(None, e)

    return reactor, requestmap, res

class queuedcommandfuture(pycompat.futures.Future):
    """Wraps result() on command futures to trigger submission on call."""

    def result(self, timeout=None):

        permission = {
            'push': 'rw',
            'pull': 'ro',
        }[permissions.pop()]

        reactor, requests, resp = sendv2request(
            self._ui, self._opener, self._requestbuilder, self._apiurl,
            permission, calls)

        # TODO we probably want to validate the HTTP code, media type, etc.

        self._responseexecutor = pycompat.futures.ThreadPoolExecutor(1)
        self._responsef = self._responseexecutor.submit(self._handleresponse,
                                                        reactor,
                                                        requests,
                                                        resp)

    def close(self):
        if self._closed:
            return

                    f.set_exception(error.ResponseError(
                        _('unfulfilled command response')))

            self._futures = None

    def _handleresponse(self, reactor, requests, resp):
        # Called in a thread to read the response.

        results = {k: [] for k in requests}

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

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

            # Guard against receiving a frame with a request ID that we
            # didn't issue. This should never happen.
            request, f = requests.get(frame.requestid, [None, None])

            action, meta = reactor.onframerecv(frame)

            if action == 'responsedata':
                assert request.requestid == meta['request'].requestid

                result = results[request.requestid]

                if meta['cbor']:
                    payload = util.bytesio(meta['data'])

                    decoder = cbor.CBORDecoder(payload)
                    while payload.tell() + 1 < len(meta['data']):
                        try:
                            result.append(decoder.decode())
                        except Exception:
                            pycompat.future_set_exception_info(
                                f, sys.exc_info()[1:])
                            continue
                else:
                    result.append(meta['data'])

                if meta['eos']:
                    f.set_result(result)
                    del results[request.requestid]

            elif action == 'error':
                e = error.RepoError(meta['message'])

                if f:
                    f.set_exception(e)
                else:
                    raise e

            else:
                e = error.ProgrammingError('unhandled action: %s' % action)

                if f:
                    f.set_exception(e)
                else:
                    raise e

# TODO implement interface for version 2 peers
@zi.implementer(repository.ipeerconnection, repository.ipeercapabilities,
                repository.ipeerrequests)
class httpv2peer(object):

37719:a656cba08a04

import errno
import io
import os
import socket
import struct
import tempfile
import weakref

from .i18n import _
from .thirdparty import (

    url as urlmod,
    util,
    wireprotoframing,
    wireprototypes,
    wireprotov1peer,
    wireprotov2peer,
    wireprotov2server,
)

httplib = util.httplib
urlerr = util.urlerr

def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests):
    reactor = wireprotoframing.clientreactor(hasmultiplesend=False,
                                             buffersends=True)

    handler = wireprotov2peer.clienthandler(ui, reactor)

    url = '%s/%s' % (apiurl, permission)

    if len(requests) > 1:
        url += '/multirequest'
    else:
        url += '/%s' % requests[0][0]

    for command, args, f in requests:
        assert not list(handler.callcommand(command, args, f))

    # TODO stream this.
    body = b''.join(map(bytes, handler.flushcommands()))

    # TODO modify user-agent to reflect v2
    headers = {
        r'Accept': wireprotov2server.FRAMINGTYPE,
        r'Content-Type': wireprotov2server.FRAMINGTYPE,

        raise
    except httplib.HTTPException as e:
        ui.traceback()
        raise IOError(None, e)

    return handler, res

class queuedcommandfuture(pycompat.futures.Future):
    """Wraps result() on command futures to trigger submission on call."""

    def result(self, timeout=None):

        permission = {
            'push': 'rw',
            'pull': 'ro',
        }[permissions.pop()]

        handler, resp = sendv2request(
            self._ui, self._opener, self._requestbuilder, self._apiurl,
            permission, calls)

        # TODO we probably want to validate the HTTP code, media type, etc.

        self._responseexecutor = pycompat.futures.ThreadPoolExecutor(1)
        self._responsef = self._responseexecutor.submit(self._handleresponse,
                                                        handler, resp)

    def close(self):
        if self._closed:
            return

                    f.set_exception(error.ResponseError(
                        _('unfulfilled command response')))

            self._futures = None

    def _handleresponse(self, handler, resp):
        # Called in a thread to read the response.

        while handler.readframe(resp):
            pass

# TODO implement interface for version 2 peers
@zi.implementer(repository.ipeerconnection, repository.ipeercapabilities,
                repository.ipeerrequests)
class httpv2peer(object):