Class JsonMemoryStream

tests/debugpy/common/test_messaging.py:26–65

Like JsonIOStream, but working directly with values stored in memory. Values are round-tripped through JSON serialization. For input, values are read from the supplied sequence or iterator. For output, values are appended to the supplied collection.

class JsonMemoryStream(object):
    """Like JsonIOStream, but working directly with values stored in memory.
    Values are round-tripped through JSON serialization.

    For input, values are read from the supplied sequence or iterator.
    For output, values are appended to the supplied collection.
    """

    json_decoder_factory = messaging.JsonIOStream.json_decoder_factory
    json_encoder_factory = messaging.JsonIOStream.json_encoder_factory

    def __init__(self, input, output, name="memory"):
        self.name = name
        self.input = iter(input)
        self.output = output

    def close(self):
        pass

    def _log_message(self, dir, data):
        format_string = "{0} {1} " + (
            "{2:indent=None}" if isinstance(data, list) else "{2}"
        )
        return log.debug(format_string, self.name, dir, json.repr(data))

    def read_json(self, decoder=None):
        decoder = decoder if decoder is not None else self.json_decoder_factory()
        try:
            value = next(self.input)
        except StopIteration:
            raise messaging.NoMoreMessages(stream=self)
        value = decoder.decode(json.dumps(value))
        self._log_message("-->", value)
        return value

    def write_json(self, value, encoder=None):
        encoder = encoder if encoder is not None else self.json_encoder_factory()
        value = json.loads(encoder.encode(value))
        self._log_message("<--", value)
        self.output.append(value)


class TestJsonIOStream(object):