/usr/share/pyshared/irc/buffer.py is in python-irc 8.5.3+dfsg-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from __future__ import unicode_literals
import re
class LineBuffer(object):
r"""
Buffer bytes read in from a connection and serve complete lines back.
>>> b = LineBuffer()
>>> len(b)
0
>>> b.feed(b'foo\nbar')
>>> len(b)
7
>>> list(b.lines()) == [b'foo']
True
>>> len(b)
3
>>> b.feed(b'bar\r\nbaz\n')
>>> list(b.lines()) == [b'barbar', b'baz']
True
>>> len(b)
0
The buffer will not perform any decoding.
>>> b.feed(b'Ol\xe9\n')
>>> list(b.lines()) == [b'Ol\xe9']
True
The LineBuffer should also act as an iterable.
>>> b.feed(b'iterate\nthis\n')
>>> for line, expected in zip(b, [b'iterate', b'this']):
... assert line == expected
"""
line_sep_exp = re.compile(b'\r?\n')
def __init__(self):
self.buffer = b''
def feed(self, bytes):
self.buffer += bytes
def lines(self):
lines = self.line_sep_exp.split(self.buffer)
# save the last, unfinished, possibly empty line
self.buffer = lines.pop()
return iter(lines)
def __iter__(self):
return self.lines()
def __len__(self):
return len(self.buffer)
class DecodingLineBuffer(LineBuffer):
r"""
Like LineBuffer, but decode the output (default assumes UTF-8).
>>> utf8_word = b'Ol\xc3\xa9'
>>> b = DecodingLineBuffer()
>>> b.feed(b'bar\r\nbaz\n' + utf8_word + b'\n')
>>> list(b.lines()) == ['bar', 'baz', utf8_word.decode('utf-8')]
True
>>> len(b)
0
Some clients will feed latin-1 or other encodings. If your client should
support docoding from these clients (and not raise a UnicodeDecodeError),
set errors='replace':
>>> rb = DecodingLineBuffer()
>>> b.errors = 'replace'
>>> b.feed(b'Ol\xe9\n')
>>> list(b.lines()) == ['Ol\ufffd']
True
"""
encoding = 'utf-8'
errors = 'strict'
def lines(self):
return (line.decode(self.encoding, self.errors)
for line in super(DecodingLineBuffer, self).lines())
|