use crate::{
Buf, OutOfBounds, RangeOutOfBounds,
python::{PyBufCmp, PyBufCommon, PyBufExt, py_check_alloc},
};
use pyo3::{
basic::CompareOp,
exceptions::PyUnicodeDecodeError,
prelude::{Bound, *},
types::{PyAny, PyBytes, PyString},
};
use std::sync::Once;
type IntoIter = ::bytes::buf::IntoIter<super::Bytes>;
const DOC: &str = r#"Bytes
Immutable shared byte buffer. Mirrors `smol_bytes.shared::Bytes`, storing up to 62 bytes inline
before falling back to heap storage with reference-counted sharing. All `Buf` getters (e.g.
`get_u16_le`) are available, along with rich comparisons, slicing, and zero-copy split helpers."#;
static DOC_ONCE: Once = Once::new();
#[derive(Debug)]
#[pyclass]
struct Iter {
inner: IntoIter,
}
#[pymethods]
impl Iter {
#[allow(clippy::self_named_constructors)]
fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
slf
}
fn __next__(mut slf: PyRefMut<'_, Self>) -> Option<u8> {
slf.inner.next()
}
}
#[pyclass(name = "Bytes", skip_from_py_object)]
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(transparent)]
pub struct PySharedBytes {
inner: super::Bytes,
}
impl From<super::Bytes> for PySharedBytes {
fn from(inner: super::Bytes) -> Self {
Self { inner }
}
}
impl AsRef<[u8]> for PySharedBytes {
#[cfg_attr(not(coverage), inline(always))]
fn as_ref(&self) -> &[u8] {
self.inner.as_ref()
}
}
impl Buf for PySharedBytes {
#[cfg_attr(not(coverage), inline(always))]
fn remaining(&self) -> usize {
self.inner.remaining()
}
#[cfg_attr(not(coverage), inline(always))]
fn chunk(&self) -> &[u8] {
self.inner.chunk()
}
#[cfg_attr(not(coverage), inline(always))]
fn advance(&mut self, cnt: usize) {
Buf::advance(&mut self.inner, cnt);
}
}
impl PyBufCommon for PySharedBytes {
#[cfg_attr(not(coverage), inline(always))]
fn py_try_split_to(&mut self, at: usize) -> Result<Self, OutOfBounds> {
self.inner.try_split_to(at).map(Into::into)
}
#[cfg_attr(not(coverage), inline(always))]
fn py_try_split_off(&mut self, at: usize) -> Result<Self, OutOfBounds> {
self.inner.try_split_off(at).map(Into::into)
}
#[cfg_attr(not(coverage), inline(always))]
fn py_try_slice(&self, start: usize, end: usize) -> Result<Self, RangeOutOfBounds> {
self.inner.try_slice(start..end).map(Into::into)
}
#[cfg_attr(not(coverage), inline(always))]
fn py_try_advance(&mut self, cnt: usize) -> Result<(), OutOfBounds> {
self.inner.try_advance(cnt)
}
}
#[pymethods]
impl PySharedBytes {
#[new]
fn new_python(py: Python<'_>) -> Self {
DOC_ONCE.call_once(|| {
let ty = py.get_type::<Self>();
let _ = ty.setattr("__doc__", DOC);
});
Self {
inner: super::Bytes::new(),
}
}
#[staticmethod]
#[pyo3(name = "from_bytes")]
fn __python_from_bytes(py_bytes: &[u8]) -> PyResult<Self> {
py_check_alloc(py_bytes.len())?;
Ok(Self {
inner: super::Bytes::copy_from_slice(py_bytes),
})
}
#[staticmethod]
#[pyo3(name = "from_str")]
fn __python_from_str(py_str: &str) -> PyResult<Self> {
py_check_alloc(py_str.len())?;
Ok(Self {
inner: super::Bytes::from(py_str),
})
}
fn __bytes__<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
self.py_bytes(py)
}
fn __bool__(&self) -> bool {
!self.inner.is_empty()
}
#[allow(non_upper_case_globals)]
const __hash__: Option<Py<PyAny>> = None;
fn __len__(&self) -> usize {
self.py_len()
}
fn __contains__(&self, item: &Bound<'_, PyAny>) -> PyResult<bool> {
self.py_contains(item)
}
fn __getitem__(&self, index: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
self.py_getitem(index)
}
fn __iter__(slf: PyRef<'_, Self>) -> PyResult<Py<Iter>> {
let iter = Iter {
inner: slf.inner.clone().into_iter(),
};
Py::new(slf.py(), iter)
}
fn __richcmp__(&self, other: &Bound<'_, PyAny>, op: CompareOp) -> PyResult<Py<PyAny>> {
self.py_richcmp(other, op)
}
fn __str__(&self, py: Python<'_>) -> PyResult<&str> {
::core::str::from_utf8(self.inner.as_ref())
.map_err(|err| PyUnicodeDecodeError::new_err_from_utf8(py, self.inner.as_ref(), err))
}
fn __repr__(&self) -> PyResult<String> {
py_check_alloc(self.inner.len().saturating_mul(6).saturating_add(64))?;
Ok(format!("{:?}", self.inner))
}
#[pyo3(name = "as_bytes")]
fn __python_as_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
self.py_bytes(py)
}
#[pyo3(name = "to_string")]
fn __python_to_string<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
self.py_to_string(py)
}
#[pyo3(name = "is_inline")]
fn __python_is_inline(&self) -> bool {
self.inner.is_inline()
}
#[pyo3(name = "is_heap")]
fn __python_is_heap(&self) -> bool {
self.inner.is_heap()
}
#[pyo3(name = "remaining")]
fn __python_remaining(&self) -> usize {
self.py_remaining()
}
#[pyo3(name = "advance")]
fn __python_advance(&mut self, cnt: usize) -> PyResult<()> {
self.py_advance(cnt)
}
#[pyo3(name = "truncate")]
fn __python_truncate(&mut self, new_len: usize) {
self.inner.truncate(new_len);
}
#[pyo3(name = "clear")]
fn __python_clear(&mut self) {
self.inner.clear();
}
#[pyo3(name = "split_to")]
fn __python_split_to(&mut self, at: usize) -> PyResult<Self> {
self.py_split_to(at)
}
#[pyo3(name = "split_off")]
fn __python_split_off(&mut self, at: usize) -> PyResult<Self> {
self.py_split_off(at)
}
#[pyo3(name = "slice")]
fn __python_slice(&self, start: usize, end: usize) -> PyResult<Self> {
self.py_slice(start, end)
}
#[pyo3(name = "__copy__")]
fn __python_copy(&self) -> Self {
self.clone()
}
#[pyo3(name = "__deepcopy__")]
fn __python_deepcopy(&self, _memo: &Bound<'_, PyAny>) -> Self {
self.clone()
}
fn __reduce__(&self, py: Python<'_>) -> PyResult<(Py<PyAny>, (Py<PyBytes>,))> {
let cls = py.get_type::<Self>();
let from_bytes = cls.getattr("from_bytes")?;
let data = PyBytes::new(py, self.inner.as_ref());
Ok((from_bytes.into(), (data.into(),)))
}
#[pyo3(name = "get_u8")]
fn __python_get_u8(&mut self) -> PyResult<u8> {
self.inner.py_get_u8()
}
#[pyo3(name = "get_i8")]
fn __python_get_i8(&mut self) -> PyResult<i8> {
self.inner.py_get_i8()
}
#[pyo3(name = "get_u16")]
fn __python_get_u16(&mut self) -> PyResult<u16> {
self.inner.py_get_u16()
}
#[pyo3(name = "get_u16_le")]
fn __python_get_u16_le(&mut self) -> PyResult<u16> {
self.inner.py_get_u16_le()
}
#[pyo3(name = "get_i16")]
fn __python_get_i16(&mut self) -> PyResult<i16> {
self.inner.py_get_i16()
}
#[pyo3(name = "get_i16_le")]
fn __python_get_i16_le(&mut self) -> PyResult<i16> {
self.inner.py_get_i16_le()
}
#[pyo3(name = "get_u32")]
fn __python_get_u32(&mut self) -> PyResult<u32> {
self.inner.py_get_u32()
}
#[pyo3(name = "get_u32_le")]
fn __python_get_u32_le(&mut self) -> PyResult<u32> {
self.inner.py_get_u32_le()
}
#[pyo3(name = "get_i32")]
fn __python_get_i32(&mut self) -> PyResult<i32> {
self.inner.py_get_i32()
}
#[pyo3(name = "get_i32_le")]
fn __python_get_i32_le(&mut self) -> PyResult<i32> {
self.inner.py_get_i32_le()
}
#[pyo3(name = "get_f32")]
fn __python_get_f32(&mut self) -> PyResult<f32> {
self.inner.py_get_f32()
}
#[pyo3(name = "get_f32_le")]
fn __python_get_f32_le(&mut self) -> PyResult<f32> {
self.inner.py_get_f32_le()
}
#[pyo3(name = "get_u64")]
fn __python_get_u64(&mut self) -> PyResult<u64> {
self.inner.py_get_u64()
}
#[pyo3(name = "get_u64_le")]
fn __python_get_u64_le(&mut self) -> PyResult<u64> {
self.inner.py_get_u64_le()
}
#[pyo3(name = "get_i64")]
fn __python_get_i64(&mut self) -> PyResult<i64> {
self.inner.py_get_i64()
}
#[pyo3(name = "get_i64_le")]
fn __python_get_i64_le(&mut self) -> PyResult<i64> {
self.inner.py_get_i64_le()
}
#[pyo3(name = "get_f64")]
fn __python_get_f64(&mut self) -> PyResult<f64> {
self.inner.py_get_f64()
}
#[pyo3(name = "get_f64_le")]
fn __python_get_f64_le(&mut self) -> PyResult<f64> {
self.inner.py_get_f64_le()
}
#[pyo3(name = "get_u128")]
fn __python_get_u128(&mut self) -> PyResult<u128> {
self.inner.py_get_u128()
}
#[pyo3(name = "get_u128_le")]
fn __python_get_u128_le(&mut self) -> PyResult<u128> {
self.inner.py_get_u128_le()
}
#[pyo3(name = "get_i128")]
fn __python_get_i128(&mut self) -> PyResult<i128> {
self.inner.py_get_i128()
}
#[pyo3(name = "get_i128_le")]
fn __python_get_i128_le(&mut self) -> PyResult<i128> {
self.inner.py_get_i128_le()
}
#[pyo3(name = "get_uint")]
fn __python_get_uint(&mut self, nbytes: &Bound<'_, PyAny>) -> PyResult<u64> {
self.inner.py_get_uint_object(nbytes)
}
#[pyo3(name = "get_uint_le")]
fn __python_get_uint_le(&mut self, nbytes: &Bound<'_, PyAny>) -> PyResult<u64> {
self.inner.py_get_uint_le_object(nbytes)
}
#[pyo3(name = "get_int")]
fn __python_get_int(&mut self, nbytes: &Bound<'_, PyAny>) -> PyResult<i64> {
self.inner.py_get_int_object(nbytes)
}
#[pyo3(name = "get_int_le")]
fn __python_get_int_le(&mut self, nbytes: &Bound<'_, PyAny>) -> PyResult<i64> {
self.inner.py_get_int_le_object(nbytes)
}
unsafe fn __getbuffer__(
slf: PyRef<'_, Self>,
view: *mut pyo3::ffi::Py_buffer,
flags: ::std::os::raw::c_int,
) -> PyResult<()> {
let py = slf.py();
let snapshot = PyBytes::new(py, slf.as_ref());
if unsafe { pyo3::ffi::PyObject_GetBuffer(snapshot.as_ptr(), view, flags) } == 0 {
Ok(())
} else {
Err(PyErr::fetch(py))
}
}
}