51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
# An iolist is one of
|
|
# - None
|
|
# - a list of iolists
|
|
# - a CountedIOList
|
|
# - a bytes
|
|
# - a number i, 0 <= i < 256
|
|
|
|
class CountedIOList:
|
|
def __init__(self, i):
|
|
self.value = i
|
|
self.length = len(i)
|
|
|
|
def counted(i):
|
|
return i if isinstance(i, CountedIOList) else CountedIOList(i)
|
|
|
|
_len = len
|
|
|
|
def len(i):
|
|
if i is None: return 0
|
|
if isinstance(i, int): return 1
|
|
if isinstance(i, _bytes): return _len(i)
|
|
if isinstance(i, list):
|
|
acc = 0
|
|
for j in i:
|
|
acc = acc + len(j)
|
|
return acc
|
|
if isinstance(i, CountedIOList): return i.length
|
|
raise ValueError('invalid iolist: ' + repr(i) + ' ' + repr(type(i)))
|
|
|
|
_bytes = bytes
|
|
|
|
def bytes(i):
|
|
buffer = bytearray(len(i))
|
|
def fill(i, offset):
|
|
while isinstance(i, CountedIOList): i = i.value
|
|
if i is None:
|
|
return offset
|
|
if isinstance(i, int):
|
|
buffer[offset] = i
|
|
return offset + 1
|
|
if isinstance(i, _bytes):
|
|
buffer[offset:offset+_len(i)] = i
|
|
return offset + _len(i)
|
|
if isinstance(i, list):
|
|
for j in i:
|
|
offset = fill(j, offset)
|
|
return offset
|
|
raise ValueError('invalid iolist')
|
|
fill(i, 0)
|
|
return _bytes(buffer)
|