tests/test-wireproto-serverreactor.py

40129:57782791b7e9

from mercurial.thirdparty import (
    cbor,
)
from mercurial import (
    util,
    wireprotoframing as framing,
)
from mercurial.utils import (
    cborutil,

ffs = framing.makeframefromhumanstring

OK = cbor.dumps({b'status': b'ok'})

def makereactor(deferoutput=False):
    return framing.serverreactor(deferoutput=deferoutput)

def sendframes(reactor, gen):
    """Send a generator of frame bytearray to a reactor.

    Emits a generator of results from ``onframerecv()`` calls.

40130:293835e0fff7

from mercurial.thirdparty import (
    cbor,
)
from mercurial import (
    ui as uimod,
    util,
    wireprotoframing as framing,
)
from mercurial.utils import (
    cborutil,

ffs = framing.makeframefromhumanstring

OK = cbor.dumps({b'status': b'ok'})

def makereactor(deferoutput=False):
    ui = uimod.ui()
    return framing.serverreactor(ui, deferoutput=deferoutput)

def sendframes(reactor, gen):
    """Send a generator of frame bytearray to a reactor.

    Emits a generator of results from ``onframerecv()`` calls.