Source code for bspump.common.bytes

from ..abc.processor import Processor


[docs] class StringToBytesParser(Processor): """ Description: | ** Default Config ** encoding : utf-8 """ ConfigDefaults = { "encoding": "utf-8", }
[docs] def __init__(self, app, pipeline, id=None, config=None): """ Description: | """ super().__init__(app, pipeline, id, config) self.Encoding = self.Config["encoding"]
[docs] def process(self, context, event): """ Description: :return: event.decode(self.Encoding) | """ assert isinstance(event, str) return event.encode(self.Encoding)
[docs] class BytesToStringParser(Processor): """ Description: | """ ConfigDefaults = { "encoding": "utf-8", }
[docs] def __init__(self, app, pipeline, id=None, config=None): """ Description: | """ super().__init__(app, pipeline, id, config) self.Encoding = self.Config["encoding"]
[docs] def process(self, context, event): """ Description: :return: event.decode(self.Encoding) | """ assert isinstance(event, bytes) return event.decode(self.Encoding)