How to use the pywatchman.bser.pdu_info function in pywatchman

To help you get started, we’ve selected a few pywatchman examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github facebook / watchman / python / pywatchman_aio / __init__.py View on Github external
async def receive(self):
        sniff = await self.transport.read(SNIFF_LEN)
        if not sniff:
            raise WatchmanError("empty watchman response")
        _1, _2, elen = bser.pdu_info(sniff)
        rlen = len(sniff)
        buf = bytearray(elen)
        buf[:rlen] = sniff
        while elen > rlen:
            b = await self.transport.read(elen - rlen)
            buf[rlen : rlen + len(b)] = b
            rlen += len(b)
        response = bytes(buf)
        try:
            res = self._loads(response)
            return res
        except ValueError as e:
            raise WatchmanError("watchman response decode error: %s" % e)