Source code for wishbone.protocol.decode.plain

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#  Copyright 2018 Jelle Smet <>
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 3 of the License, or
#  (at your option) any later version.
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  GNU General Public License for more details.
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
#  MA 02110-1301, USA.

from wishbone.protocol import Decode
from io import StringIO
import os

[docs]class Plain(Decode): ''' Decode string data into a Python object. Parameters: - charset(str)('utf-8) | The charset to use when decoding. - delimiter(str)(None) | The dilimiter to use in a stream of Python docs. | None expects the complete payload to be 1 json document. - buffer_size(int)(4096) | The maximum number of bytes allowed to buffer. - strip_newline(bool)(False) | If ``True`` strips newline from each line. ''' def __init__(self, charset='utf-8', delimiter=None, buffer_size=4096, strip_newline=False): self.charset = charset if isinstance(delimiter, bytes): self.delimiter = delimiter.decode(charset, 'strict') else: self.delimiter = delimiter self.buffer_size = buffer_size self.strip_newline = strip_newline self.buffer = StringIO() def getBufferSize(self): cur_pos = self.buffer.tell(), os.SEEK_END) size = self.buffer.tell() return size def handleBytes(self, data): for chunk in self.handler(data.decode(self.charset, "strict")): yield chunk def handleDict(self, data): return data def handleReadLinesMethod(self, data): for chunk in data.readlines(): for item in self.handler(chunk): yield item def handleString(self, data): self.buffer.write(data) if self.delimiter is not None and self.delimiter in self.buffer.getvalue(): for item in self.buffer.getvalue().split(self.delimiter): if item == "": next else: if self.strip_newline: yield item.rstrip() else: yield item self.resetBuffer() else: def flush(self): if self.getBufferSize() == 0: return yield else: if self.strip_newline: yield self.buffer.getvalue().rstrip() else: yield self.buffer.getvalue() self.resetBuffer() def resetBuffer(self): self.buffer = StringIO()