# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

cimport cython

from cpython.object cimport PyObject

from cpython.buffer cimport (

Py_buffer,

PyObject_GetBuffer,

PyBuffer_Release,

PyBUF_SIMPLE,

PyBUF_WRITABLE,

)

from cpython.unicode cimport (

PyUnicode_GET_LENGTH,

PyUnicode_KIND,

PyUnicode_1BYTE_KIND,

PyUnicode_2BYTE_KIND,

PyUnicode_DATA,

PyUnicode_AsUTF8AndSize,

PyUnicode_DecodeLatin1,

PyUnicode_DecodeUTF16,

PyUnicode_FromKindAndData,

PyUnicode_DecodeUTF8,

)

from cpython.bytes cimport PyBytes_FromStringAndSize, PyBytes_AS_STRING, PyBytes_GET_SIZE

from libcpp.memory cimport shared_ptr, unique_ptr

from libcpp.utility cimport move

from cython.operator cimport dereference as deref

from libcpp.string cimport string as c_string

from libc.string cimport memcpy

from libc.stdint cimport *

from libcpp cimport bool as c_bool

from pyfory.includes.libutil cimport(

CBuffer, COutputStream, allocate_buffer, get_bit as c_get_bit, set_bit as c_set_bit, clear_bit as c_clear_bit,

set_bit_to as c_set_bit_to, CError, CErrorCode, CResultVoidError, utf16_has_surrogate_pairs

)

from pyfory.includes.libpyfory cimport (

Fory_PyCreateBufferFromStream,

Fory_PyCreateOutputStream,

Fory_PyBindBufferToOutputStream,

Fory_PyClearBufferOutputStream

)

from pyfory.error import raise_fory_error

cdef int32_t max_buffer_size = 2 ** 31 - 1

cdef int UTF16_LE = -1

@cython.final

cdef class _SharedBufferOwner:

cdef shared_ptr[CBuffer] buffer

cdef class Buffer

@cython.final

cdef class PyOutputStream:

cdef object stream

cdef unique_ptr[COutputStream] c_output_stream

@staticmethod

cdef inline PyOutputStream from_stream(object stream):

cdef c_string stream_error

cdef COutputStream* raw_writer = NULL

if stream is None:

raise ValueError("stream must not be None")

if Fory_PyCreateOutputStream(

<PyObject*>stream, &raw_writer, &stream_error

) != 0:

raise ValueError(stream_error.decode("UTF-8"))

cdef PyOutputStream writer = PyOutputStream.__new__(PyOutputStream)

writer.stream = stream

writer.c_output_stream.reset(raw_writer)

if raw_writer != NULL:

raw_writer.reset()

return writer

cdef inline COutputStream* get_c_output_stream(self):

return self.c_output_stream.get()

cpdef inline object get_output_stream(self):

return self.stream

cpdef inline void reset(self):

cdef COutputStream* output_stream = self.c_output_stream.get()

if output_stream == NULL:

raise ValueError("OutputStream is null")

output_stream.reset()

cpdef inline void enter_flush_barrier(self):

cdef COutputStream* output_stream = self.c_output_stream.get()

if output_stream == NULL:

raise ValueError("OutputStream is null")

output_stream.enter_flush_barrier()

cpdef inline void exit_flush_barrier(self):

cdef COutputStream* output_stream = self.c_output_stream.get()

if output_stream == NULL:

raise ValueError("OutputStream is null")

output_stream.exit_flush_barrier()

cpdef inline void try_flush(self):

cdef COutputStream* output_stream = self.c_output_stream.get()

if output_stream == NULL:

raise ValueError("OutputStream is null")

output_stream.try_flush()

if output_stream.has_error():

raise ValueError(output_stream.error().to_string().decode("UTF-8"))

cpdef inline void force_flush(self):

cdef COutputStream* output_stream = self.c_output_stream.get()

if output_stream == NULL:

raise ValueError("OutputStream is null")

output_stream.force_flush()

if output_stream.has_error():

raise ValueError(output_stream.error().to_string().decode("UTF-8"))

cpdef inline PyOutputStream _wrap_output_stream(object stream):

return PyOutputStream.from_stream(stream)

@cython.final

cdef class Buffer:

cdef:

unique_ptr[CBuffer] c_buffer_owner

CBuffer* c_buffer

CError _error

# hold python buffer reference count

object data

object output_stream

c_bool readonly

int32_t export_count

Py_ssize_t shape[1]

Py_ssize_t stride[1]

def __init__(self, data not None, int32_t offset=0, length=None):

cdef object view = None

cdef bint bytes_input = type(data) is bytes

cdef Py_ssize_t buffer_len

cdef Py_ssize_t py_length

cdef int32_t length_

cdef Py_buffer py_buffer

if bytes_input:

buffer_len = PyBytes_GET_SIZE(data)

else:

view = memoryview(data)

buffer_len = view.nbytes

if buffer_len > max_buffer_size:

raise ValueError(f"Buffer size {buffer_len} exceeds the maximum supported size")

if length is None:

py_length = buffer_len - offset

else:

py_length = length

if offset < 0 or offset > buffer_len or py_length < 0 or py_length > buffer_len - offset:

raise ValueError(f'Wrong offset {offset} or length {length} for buffer with size {buffer_len}')

length_ = <int32_t>py_length

cdef uint8_t* address

if length_ > 0:

if bytes_input:

address = <uint8_t*>PyBytes_AS_STRING(data) + offset

else:

if PyObject_GetBuffer(view, &py_buffer, PyBUF_SIMPLE) != 0:

raise BufferError(f"Cannot access buffer for {type(data)!r}")

try:

address = <uint8_t*>py_buffer.buf + offset

finally:

PyBuffer_Release(&py_buffer)

else:

address = NULL

if bytes_input:

self.data = data

self.readonly = True

else:

# Retaining the memoryview keeps resizable exporters locked while the native view exists.

self.data = view

self.readonly = view.readonly

self.c_buffer_owner.reset(new CBuffer(address, length_, False))

self.c_buffer = self.c_buffer_owner.get()

self.c_buffer.reader_index(0)

self.c_buffer.writer_index(0)

self.output_stream = None

self.export_count = 0

@classmethod

def from_stream(cls, stream not None, uint32_t buffer_size=4096):

cdef CBuffer* stream_buffer

cdef c_string stream_error

if buffer_size > <uint32_t>max_buffer_size:

raise ValueError(f"Buffer size {buffer_size} exceeds the maximum supported size")

if Fory_PyCreateBufferFromStream(

<PyObject*>stream, buffer_size, &stream_buffer, &stream_error

) != 0:

raise ValueError(stream_error.decode("UTF-8"))

if stream_buffer == NULL:

raise ValueError("failed to create stream buffer")

cdef Buffer buffer = Buffer.__new__(Buffer)

buffer.c_buffer_owner.reset(stream_buffer)

buffer.c_buffer = buffer.c_buffer_owner.get()

buffer.data = stream

buffer.output_stream = None

buffer.readonly = False

buffer.export_count = 0

buffer.c_buffer.reader_index(0)

buffer.c_buffer.writer_index(0)

return buffer

@staticmethod

cdef Buffer wrap(shared_ptr[CBuffer] c_buffer):

cdef Buffer buffer = Buffer.__new__(Buffer)

cdef CBuffer* ptr = c_buffer.get()

buffer.c_buffer = ptr

cdef _SharedBufferOwner owner = _SharedBufferOwner.__new__(_SharedBufferOwner)

owner.buffer = c_buffer

buffer.data = owner

buffer.output_stream = None

buffer.readonly = False

buffer.export_count = 0

buffer.c_buffer.reader_index(0)

buffer.c_buffer.writer_index(0)

return buffer

@classmethod

def allocate(cls, int32_t size):

if size < 0:

raise ValueError("Buffer size must be non-negative")

cdef CBuffer* buf = allocate_buffer(size)

if buf == NULL:

raise MemoryError("out of memory")

cdef Buffer buffer = Buffer.__new__(Buffer)

buffer.c_buffer_owner.reset(buf)

buffer.c_buffer = buffer.c_buffer_owner.get()

buffer.data = None

buffer.output_stream = None

buffer.readonly = False

buffer.export_count = 0

buffer.c_buffer.reader_index(0)

buffer.c_buffer.writer_index(0)

return buffer

@staticmethod

def wrap_output_stream(stream):

return _wrap_output_stream(stream)

cpdef inline void bind_output_stream(self, object output):

cdef c_string stream_error

cdef PyOutputStream output_stream

if Fory_PyClearBufferOutputStream(self.c_buffer, &stream_error) != 0:

raise ValueError(stream_error.decode("UTF-8"))

if output is None:

self.output_stream = None

return

if isinstance(output, PyOutputStream):

output_stream = <PyOutputStream>output

else:

output_stream = _wrap_output_stream(output)

output_stream.reset()

if Fory_PyBindBufferToOutputStream(

self.c_buffer, output_stream.get_c_output_stream(), &stream_error

) != 0:

raise ValueError(stream_error.decode("UTF-8"))

self.output_stream = output_stream

cpdef inline object get_output_stream(self):

return self.output_stream

cdef inline void _raise_if_error(self):

cdef CErrorCode code

cdef c_string message

if not self._error.ok():

code = self._error.code()

message = self._error.message()

self._error.reset()

raise_fory_error(code, message)

cdef inline void _check_writable(self):

if self.readonly:

raise TypeError("cannot modify read-only Buffer")

cdef inline int32_t _checked_length(self, Py_ssize_t length, str owner):

if length < 0 or length > max_buffer_size:

raise ValueError(f"{owner} length {length} exceeds the maximum supported size")

return <int32_t>length

cdef inline void _check_resize(self, int32_t new_size):

if new_size > self.c_buffer.size() and self.export_count != 0:

raise BufferError("cannot resize Buffer while exported views exist")

cdef inline void _prepare_capacity(self, int32_t needed_size):

cdef int32_t writer_index = <int32_t>self.c_buffer.writer_index()

cdef int32_t end

if needed_size < 0 or needed_size > max_buffer_size - writer_index:

raise ValueError(

f"Buffer growth {needed_size} exceeds the maximum supported size at writer index "

f"{self.c_buffer.writer_index()}"

)

end = writer_index + needed_size

if end > self.c_buffer.size():

self._check_resize(end)

if end > max_buffer_size // 2:

self.c_buffer.reserve(<uint32_t>end)

cdef inline void _prepare_write(self, int32_t needed_size):

self._check_writable()

self._prepare_capacity(needed_size)

cdef inline void _grow_checked(self, int32_t needed_size):

self._prepare_write(needed_size)

self.c_buffer.grow(<uint32_t>needed_size)

cpdef inline int32_t get_reader_index(self):

return <int32_t>self.c_buffer.reader_index()

cpdef inline void set_reader_index(self, int32_t value):

if value < 0:

raise ValueError("reader_index must be >= 0")

if not self.c_buffer.reader_index(<uint32_t>value, self._error):

if not self._error.ok():

self._raise_if_error()

cpdef inline int32_t get_writer_index(self):

return <int32_t>self.c_buffer.writer_index()

cpdef inline void set_writer_index(self, int32_t value):

if value < 0 or value > self.c_buffer.size():

raise ValueError(f"writer_index {value} out of bound {0, self.c_buffer.size()}")

self.c_buffer.writer_index(<uint32_t>value)

cpdef inline void shrink_input_buffer(self):

self.c_buffer.shrink_input_buffer()

cpdef inline c_bool has_input_stream(self):

return self.c_buffer.has_input_stream()

cpdef c_bool own_data(self):

return self.c_buffer.own_data()

cpdef inline reserve(self, int32_t new_size):

if new_size < 0:

raise ValueError(f"Buffer size {new_size} out of bound {0, max_buffer_size}")

self._check_resize(new_size)

self.c_buffer.reserve(new_size)

cpdef inline put_bool(self, uint32_t offset, c_bool v):

self._check_writable()

self.check_bound(offset, <int32_t>1)

self.c_buffer.unsafe_put_byte(offset, v)

cpdef inline put_uint8(self, uint32_t offset, uint8_t v):

self._check_writable()

self.check_bound(offset, <int32_t>1)

self.c_buffer.unsafe_put_byte(offset, v)

cpdef inline put_int8(self, uint32_t offset, int8_t v):

self._check_writable()

self.check_bound(offset, <int32_t>1)

self.c_buffer.unsafe_put_byte(offset, v)

cpdef inline put_int16(self, uint32_t offset, int16_t v):

self._check_writable()

self.check_bound(offset, <int32_t>2)

self.c_buffer.unsafe_put(offset, v)

cpdef inline put_int24(self, uint32_t offset, int32_t v):

self._check_writable()

self.check_bound(offset, <int32_t>3)

self.c_buffer.put_int24(offset, v)

cpdef inline put_int32(self, uint32_t offset, int32_t v):

self._check_writable()

self.check_bound(offset, <int32_t>4)

self.c_buffer.unsafe_put(offset, v)

cpdef inline put_int64(self, uint32_t offset, int64_t v):

self._check_writable()

self.check_bound(offset, <int32_t>8)

self.c_buffer.unsafe_put(offset, v)

cpdef inline put_float(self, uint32_t offset, float v):

self._check_writable()

self.check_bound(offset, <int32_t>4)

self.c_buffer.unsafe_put(offset, v)

cpdef inline put_double(self, uint32_t offset, double v):

self._check_writable()

self.check_bound(offset, <int32_t>8)

self.c_buffer.unsafe_put(offset, v)

cpdef inline c_bool get_bool(self, uint32_t offset):

self.check_bound(offset, <int32_t>1)

return self.c_buffer.get_bool(offset)

cpdef inline int8_t get_int8(self, uint32_t offset):

self.check_bound(offset, <int32_t>1)

return self.c_buffer.get_int8(offset)

cpdef inline int16_t get_int16(self, uint32_t offset):

self.check_bound(offset, <int32_t>2)

return self.c_buffer.get_int16(offset)

cpdef inline int32_t get_int24(self, uint32_t offset):

self.check_bound(offset, <int32_t>3)

cdef int32_t value = self.c_buffer.get_int24(offset)

if value & 0x800000:

value -= 0x1000000

return value

cpdef inline int32_t get_int32(self, uint32_t offset):

self.check_bound(offset, <int32_t>4)

return self.c_buffer.get_int32(offset)

cpdef inline int64_t get_int64(self, uint32_t offset):

self.check_bound(offset, <int32_t>8)

return self.c_buffer.get_int64(offset)

cpdef inline float get_float(self, uint32_t offset):

self.check_bound(offset, <int32_t>4)

return self.c_buffer.get_float(offset)

cpdef inline double get_double(self, uint32_t offset):

self.check_bound(offset, <int32_t>8)

return self.c_buffer.get_double(offset)

cpdef inline check_bound(self, int32_t offset, int32_t length):

cdef int32_t size_ = self.c_buffer.size()

if offset < 0 or length < 0 or offset > size_ or length > size_ - offset:

raise_fory_error(

CErrorCode.BufferOutOfBound,

f"Address range {offset, offset + length} out of bound {0, size_}",

)

cpdef inline ensure_readable(self, int32_t length):

if length < 0:

raise_fory_error(CErrorCode.InvalidData, f"Readable byte count {length} is negative")

if length == 0:

return

if not self.c_buffer.ensure_readable(<uint32_t>length, self._error):

if not self._error.ok():

self._raise_if_error()

cpdef inline write_bool(self, c_bool value):

self._prepare_write(1)

self.c_buffer.write_uint8(<uint8_t>value)

cpdef inline write_uint8(self, uint8_t value):

self._prepare_write(1)

self.c_buffer.write_uint8(value)

cpdef inline write_int8(self, int8_t value):

self._prepare_write(1)

self.c_buffer.write_int8(value)

cpdef inline write_int16(self, int16_t value):

self._prepare_write(2)

self.c_buffer.write_int16(value)

cpdef inline write_int24(self, int32_t value):

self._prepare_write(3)

self.c_buffer.write_int24(value)

cpdef inline write_int32(self, int32_t value):

self._prepare_write(4)

self.c_buffer.write_int32(value)

cpdef inline write_int64(self, int64_t value):

self._prepare_write(8)

self.c_buffer.write_int64(value)

cpdef inline write_uint16(self, uint16_t value):

self._prepare_write(2)

self.c_buffer.write_uint16(value)

cpdef inline write_uint32(self, uint32_t value):

self._prepare_write(4)

self.c_buffer.write_uint32(value)

cpdef inline write_uint64(self, uint64_t value):

self._prepare_write(8)

self.c_buffer.write_int64(<int64_t>value)

cpdef inline write_float(self, float value):

self._prepare_write(4)

self.c_buffer.write_float(value)

cpdef inline write_float32(self, float value):

self._prepare_write(4)

self.c_buffer.write_float(value)

cpdef inline write_double(self, double value):

self._prepare_write(8)

self.c_buffer.write_double(value)

cpdef inline write_float64(self, double value):

self._prepare_write(8)

self.c_buffer.write_double(value)

cdef inline void _buffer_byte_range(

self,

object view,

Py_ssize_t src_index,

Py_ssize_t length,

int32_t* src_offset,

int32_t* size,

):

cdef Py_ssize_t itemsize = view.itemsize

cdef Py_ssize_t element_count

cdef Py_ssize_t src_offset_

cdef Py_ssize_t size_

if not view.c_contiguous:

raise ValueError("Source buffer must be C-contiguous")

if itemsize <= 0:

raise ValueError(f"Invalid source buffer item size {itemsize}")

element_count = view.nbytes // itemsize

if src_index < 0 or src_index > element_count or length < 0 or length > element_count - src_index:

raise ValueError(f"Source range {(src_index, length)} out of bound {(0, element_count)}")

src_offset_ = src_index * itemsize

size_ = length * itemsize

if src_offset_ > max_buffer_size or size_ > max_buffer_size:

raise ValueError("Source byte range exceeds the maximum supported size")

src_offset[0] = <int32_t>src_offset_

size[0] = <int32_t>size_

cdef inline void _copy_buffer(self, uint32_t offset, object view, int32_t src_offset, int32_t size):

cdef Py_buffer py_buffer

self.check_bound(<int32_t>offset, size)

if size == 0:

return

if PyObject_GetBuffer(view, &py_buffer, PyBUF_SIMPLE) != 0:

raise BufferError(f"Cannot access buffer for {type(view)!r}")

try:

self.c_buffer.copy_from(offset, <uint8_t*>py_buffer.buf, src_offset, size)

finally:

PyBuffer_Release(&py_buffer)

cpdef put_buffer(self, uint32_t offset, v, Py_ssize_t src_index, Py_ssize_t length):

cdef object view = memoryview(v)

cdef int32_t src_offset

cdef int32_t size

self._buffer_byte_range(view, src_index, length, &src_offset, &size)

if size > 0:

self._check_writable()

self._copy_buffer(offset, view, src_offset, size)

cpdef inline write_bytes_and_size(self, bytes value):

cdef int32_t length = self._checked_length(PyBytes_GET_SIZE(value), "Binary")

self.write_var_uint32(length)

if length > 0:

self._prepare_write(length)

self.c_buffer.write_bytes(PyBytes_AS_STRING(value), length)

cpdef inline bytes read_bytes_and_size(self):

cdef int32_t length = self.read_var_uint32()

return self.read_bytes(length)

cpdef inline write_bytes(self, bytes value):

cdef int32_t length = self._checked_length(PyBytes_GET_SIZE(value), "Binary")

if length > 0:

self._prepare_write(length)

self.c_buffer.write_bytes(PyBytes_AS_STRING(value), length)

cpdef inline bytes read_bytes(self, int32_t length):

if length == 0:

return b""

if length < 0:

raise_fory_error(CErrorCode.InvalidData, f"Binary size {length} is negative")

if not self.c_buffer.ensure_readable(<uint32_t>length, self._error):

if not self._error.ok():

self._raise_if_error()

cdef bytes py_bytes = PyBytes_FromStringAndSize(NULL, length)

if py_bytes is None:

raise MemoryError("out of memory")

cdef char* buf = PyBytes_AS_STRING(py_bytes)

cdef uint32_t offset = self.c_buffer.reader_index()

memcpy(buf, self.c_buffer.data() + offset, <size_t>length)

self.c_buffer.reader_index(offset + <uint32_t>length)

return py_bytes

cpdef inline int64_t read_bytes_as_int64(self, int32_t length):

cdef int64_t result = 0

cdef uint32_t offset

cdef CResultVoidError res

if length == 0:

return 0

if length < 0 or length > 8:

raise_fory_error(

CErrorCode.InvalidData,

f"get_bytes_as_int64 length should be in range [0, 8], but got {length}",

)

if not self.c_buffer.ensure_readable(<uint32_t>length, self._error):

if not self._error.ok():

self._raise_if_error()

offset = self.c_buffer.reader_index()

res = self.c_buffer.get_bytes_as_int64(offset, <uint32_t>length, &result)

if not res.ok():

raise_fory_error(res.error().code(), res.error().message())

self.c_buffer.increase_reader_index(<uint32_t>length, self._error)

if not self._error.ok():

self._raise_if_error()

return result

cpdef inline put_bytes(self, uint32_t offset, bytes value):

cdef const unsigned char[:] data = value

cdef int32_t length = self._checked_length(data.nbytes, "Binary")

cdef uint64_t end = <uint64_t>offset + <uint64_t>length

if length > 0:

self._check_writable()

if end > <uint64_t>max_buffer_size:

raise ValueError(f"Destination range {(offset, end)} exceeds the maximum supported size")

if end > <uint64_t>self.c_buffer.size():

self.reserve(<int32_t>end)

self.c_buffer.copy_from(offset, &data[0], 0, length)

cpdef inline bytes get_bytes(self, uint32_t offset, uint32_t nbytes):

if nbytes == 0:

return b""

self.check_bound(offset, nbytes)

cdef unsigned char* binary_data = self.c_buffer.data() + offset

return binary_data[:nbytes]

cpdef inline write_buffer(self, value, src_index=0, length_=None):

cdef object view = memoryview(value)

cdef Py_ssize_t itemsize = view.itemsize

cdef Py_ssize_t element_count

cdef Py_ssize_t src_index_ = src_index

cdef Py_ssize_t length

cdef int32_t src_offset

cdef int32_t size

if itemsize <= 0:

raise ValueError(f"Invalid source buffer item size {itemsize}")

element_count = view.nbytes // itemsize

if length_ is None:

length = element_count - src_index_

else:

length = length_

self._buffer_byte_range(view, src_index_, length, &src_offset, &size)

if size == 0:

return

cdef uint32_t offset = self.c_buffer.writer_index()

self._grow_checked(size)

self._copy_buffer(offset, view, src_offset, size)

self.c_buffer.increase_writer_index(size)

cpdef inline write(self, value):

cdef const unsigned char[::1] data

cdef int32_t length

data = value

length = self._checked_length(data.nbytes, "Buffer")

if length > 0:

self._prepare_write(length)

self.c_buffer.write_bytes(&data[0], length)

cpdef inline grow(self, int32_t needed_size):

self._prepare_capacity(needed_size)

self.c_buffer.grow(<uint32_t>needed_size)

cpdef inline ensure(self, int32_t length):

if length < 0:

raise ValueError("Buffer size must be non-negative")

if length > self.c_buffer.size():

self.reserve(length)

cpdef inline skip(self, uint32_t length):

self.c_buffer.skip(length, self._error)

if not self._error.ok():

self._raise_if_error()

cpdef inline c_bool read_bool(self):

cdef uint8_t value = self.c_buffer.read_uint8(self._error)

if not self._error.ok():

self._raise_if_error()

return value != 0

cpdef inline uint8_t read_uint8(self):

cdef uint8_t value = self.c_buffer.read_uint8(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline int8_t read_int8(self):

cdef int8_t value = self.c_buffer.read_int8(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline int16_t read_int16(self):

cdef int16_t value = self.c_buffer.read_int16(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline int32_t read_int24(self):

cdef int32_t value = self.c_buffer.read_int24(self._error)

if not self._error.ok():

self._raise_if_error()

if value & 0x800000:

value -= 0x1000000

return value

cpdef inline int32_t read_int32(self):

cdef int32_t value = self.c_buffer.read_int32(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline int64_t read_int64(self):

cdef int64_t value = self.c_buffer.read_int64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline uint16_t read_uint16(self):

cdef uint16_t value = self.c_buffer.read_uint16(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline uint32_t read_uint32(self):

cdef uint32_t value = self.c_buffer.read_uint32(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline uint64_t read_uint64(self):

cdef uint64_t value = self.c_buffer.read_uint64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline float read_float(self):

cdef float value = self.c_buffer.read_float(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline float read_float32(self):

cdef float value = self.c_buffer.read_float(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline double read_double(self):

cdef double value = self.c_buffer.read_double(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline double read_float64(self):

cdef double value = self.c_buffer.read_double(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline bytes read(self, int32_t length):

return self.read_bytes(length)

cpdef inline bytes readline(self, int32_t size=-1):

if size != <int32_t>-1:

raise ValueError(f"Specify size {size} is unsupported")

cdef uint8_t* arr = self.c_buffer.data()

cdef uint32_t start_index = self.c_buffer.reader_index()

cdef uint32_t target_index = start_index

cdef uint8_t sep = 10 # '\n'

cdef int32_t buffer_size = self.c_buffer.size()

while target_index < <uint32_t>buffer_size and arr[target_index] != sep:

target_index += <int32_t>1

cdef bytes data = arr[start_index:target_index]

self.c_buffer.reader_index(target_index)

return data

cpdef inline write_varint32(self, int32_t value):

self._prepare_write(8)

cdef uint32_t before = self.c_buffer.writer_index()

self.c_buffer.write_var_int32(value)

cdef uint32_t after = self.c_buffer.writer_index()

return after - before

cpdef inline write_var_uint32(self, uint32_t value):

self._prepare_write(8)

cdef uint32_t before = self.c_buffer.writer_index()

self.c_buffer.write_var_uint32(value)

cdef uint32_t after = self.c_buffer.writer_index()

return after - before

cpdef inline int32_t read_varint32(self):

cdef int32_t value = self.c_buffer.read_var_int32(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline uint32_t read_var_uint32(self):

cdef uint32_t value = self.c_buffer.read_var_uint32(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline write_varint64(self, int64_t value):

self._prepare_write(9)

cdef uint32_t before = self.c_buffer.writer_index()

self.c_buffer.write_var_int64(value)

cdef uint32_t after = self.c_buffer.writer_index()

return after - before

cpdef inline write_var_uint64(self, uint64_t v):

self._prepare_write(9)

cdef uint32_t before = self.c_buffer.writer_index()

self.c_buffer.write_var_uint64(v)

cdef uint32_t after = self.c_buffer.writer_index()

return after - before

cpdef inline int64_t read_varint64(self):

cdef int64_t value = self.c_buffer.read_var_int64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline uint64_t read_var_uint64(self):

cdef uint64_t value = self.c_buffer.read_var_uint64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline write_tagged_int64(self, int64_t value):

"""write signed int64 using fory Tagged(Small long as int) encoding."""

self._prepare_write(9)

self.c_buffer.write_tagged_int64(value)

cpdef inline int64_t read_tagged_int64(self):

"""Read signed fory Tagged(Small long as int) encoded int64."""

cdef int64_t value = self.c_buffer.read_tagged_int64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cpdef inline write_tagged_uint64(self, uint64_t value):

"""write unsigned uint64 using fory Tagged(Small long as int) encoding."""

self._prepare_write(9)

self.c_buffer.write_tagged_uint64(value)

cpdef inline uint64_t read_tagged_uint64(self):

"""Read unsigned fory Tagged(Small long as int) encoded uint64."""

cdef uint64_t value = self.c_buffer.read_tagged_uint64(self._error)

if not self._error.ok():

self._raise_if_error()

return value

cdef inline write_c_buffer(self, const uint8_t* value, int32_t length):

self.write_var_uint32(length)

if length <= 0: # access an emtpy buffer may raise out-of-bound exception.

return

cdef uint32_t offset = self.c_buffer.writer_index()

self._grow_checked(length)

self.check_bound(offset, length)

self.c_buffer.copy_from(offset, value, 0, length)

self.c_buffer.increase_writer_index(length)

cpdef inline write_string(self, str value):

cdef Py_ssize_t length = PyUnicode_GET_LENGTH(value)

cdef int32_t kind = PyUnicode_KIND(value)

# Note: buffer will be native endian for PyUnicode_2BYTE_KIND

cdef void* buffer = PyUnicode_DATA(value)

cdef uint64_t header = 0

cdef int32_t buffer_size

if kind == PyUnicode_1BYTE_KIND:

buffer_size = self._checked_length(length, "String")

header = (length << 2) | 0

elif kind == PyUnicode_2BYTE_KIND:

if length > max_buffer_size >> 1:

raise ValueError(f"String length {length} exceeds the maximum supported size")

buffer_size = <int32_t>(length << 1)

header = (length << 3) | 1

else:

buffer = <void *>(PyUnicode_AsUTF8AndSize(value, &length))

buffer_size = self._checked_length(length, "UTF-8 string")

header = (buffer_size << 2) | 2

self.write_var_uint64(header)

if buffer_size == 0: # access an emtpy buffer may raise out-of-bound exception.

return

cdef uint32_t offset = self.c_buffer.writer_index()

self._grow_checked(buffer_size)

self.check_bound(offset, buffer_size)

self.c_buffer.copy_from(offset, <const uint8_t *>buffer, 0, buffer_size)

self.c_buffer.increase_writer_index(buffer_size)

cpdef inline str read_string(self):

cdef uint64_t header = self.read_var_uint64()

cdef uint64_t size64 = header >> 2

if size64 > <uint64_t>2147483647:

raise ValueError(f"String size {size64} exceeds the maximum supported size")

cdef uint32_t size = <uint32_t>size64

cdef uint32_t encoding = header & <uint32_t>0b11

if size == 0:

return ""

cdef uint32_t offset = self.c_buffer.reader_index()

cdef uint32_t available = self.c_buffer.size() - offset

cdef const char * buf

cdef bytes py_bytes

if available >= size:

buf = <const char *>(self.c_buffer.data() + offset)

self.c_buffer.reader_index(offset + size)

else:

py_bytes = self.read_bytes(<int32_t>size)

buf = <const char *>PyBytes_AS_STRING(py_bytes)

if encoding == 0:

# PyUnicode_FromASCII

return PyUnicode_DecodeLatin1(buf, size, "strict")

elif encoding == 1:

if utf16_has_surrogate_pairs(<const uint16_t *>buf, size >> 1):

return PyUnicode_DecodeUTF16(

buf,

size, # len of string in bytes

NULL, # special error handling options, we don't need any

&UTF16_LE, # fory use little-endian

)

else:

return PyUnicode_FromKindAndData(PyUnicode_2BYTE_KIND, buf, size >> 1)

else:

return PyUnicode_DecodeUTF8(buf, size, "strict")

def __len__(self):

return self.c_buffer.size()

cpdef inline int32_t size(self):

return self.c_buffer.size()

def to_bytes(self, int32_t offset=0, int32_t length=0) -> bytes:

cdef int32_t size_ = self.c_buffer.size()

if offset < 0 or offset > size_:

raise ValueError(f"offset {offset} out of bound {0, size_}")

if length < 0:

raise ValueError(f"length {length} must be non-negative")

if length == 0:

length = size_ - offset

elif length > size_ - offset:

raise ValueError(f"Address range {(offset, offset + length)} out of bound {(0, size_)}")

cdef:

uint8_t* data = self.c_buffer.data() + offset

return data[:length]

def to_pybytes(self) -> bytes:

return self.to_bytes()

def slice(self, offset=0, length=None):

return type(self)(self, offset, length)

def __getitem__(self, key):

if isinstance(key, slice):

if (key.step or 1) != 1:

raise IndexError('only slices with step 1 supported')

return _normalize_slice(self, key)

return self.getitem(_normalize_index(key, self.c_buffer.size()))

cdef getitem(self, int64_t i):

return self.c_buffer.data()[i]

def hex(self):

"""

Compute hexadecimal representation of the buffer.

Returns

-------

: bytes

"""

return self.c_buffer.hex().decode("UTF-8")

def __getbuffer__(self, Py_buffer *buffer, int flags):

cdef Py_ssize_t itemsize = 1

if self.c_buffer.has_input_stream():

raise BufferError("stream-backed Buffer cannot export a relocatable view")

if self.readonly and flags & PyBUF_WRITABLE:

raise BufferError("Buffer is read-only")

self.shape[0] = self.c_buffer.size()

self.stride[0] = itemsize

buffer.buf = <char *>(self.c_buffer.data())

buffer.format = 'B'

buffer.internal = NULL # see References

buffer.itemsize = itemsize

buffer.len = self.c_buffer.size() # product(shape) * itemsize

buffer.ndim = 1

buffer.obj = self

buffer.readonly = self.readonly

buffer.shape = self.shape

buffer.strides = self.stride

buffer.suboffsets = NULL # for pointer arrays only

self.export_count += 1