slbotbm commented on code in PR #3613:
URL: https://github.com/apache/iggy/pull/3613#discussion_r3565819049
##########
foreign/python/src/user_headers.rs:
##########
Review Comment:
Some comments:
- I'd like to stick to as much idiomatic rust as possible, so please convert
the `to_rust` and `from_rust` functions to `TryFrom` impls. You could declare
something like `struct RustHeaderKeyRef<'py, 'value> {py: Python<'py>, value:
&'value RustHeaderKey}` so that you can covert `PyBytes` to rust
representation. You'll need to define 4 of these for the conversion.
- Also, please inline as much code is as possible since I'd like to prevent
the usage of ad-hoc functions created just for a small convenience. I generally
create ad-hoc functions if there is scope to utilize it in the whole sdk.
Otherwise it will keep on accumulating.
##########
examples/python/message-headers/consumer.py:
##########
Review Comment:
Instead of duplicating functions and enums in consumer.py and producer.py,
you can define a common.py file and move all common code there.
Maybe also consider splitting the examples into 4 files: producer/consumer
for typed headers, and producer/consumer for normal headers. That, or
demonstrate only typed headers, with a comment stating that normal pythonic
headers are also possible and show what kind. something like the following:
```
message-headers/
├── message-type/
│ ├── common.py
│ ├── producer.py
│ └── consumer.py
└── typed-headers/
├── producer.py
└── consumer.py
```
Also, add return types for all functions. Some are missing them
##########
examples/python/message-headers/consumer.py:
##########
@@ -0,0 +1,170 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any, NamedTuple
+
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+ UserHeaders,
+)
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy client, e.g. "
+ "'iggy+tcp://iggy:[email protected]:8090'"
+ ),
+ default="iggy+tcp://iggy:[email protected]:8090",
+ nargs="?",
+ type=str,
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def main():
+ args: ArgNamespace = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+ logger.info("Connecting to Iggy")
+ await client.connect()
+ logger.info("Connected")
+ await consume_messages(client)
+
+
+async def consume_messages(client: IggyClient):
+ interval = 0.5
+ logger.info(
+ f"Messages will be consumed from stream: {STREAM_NAME}, "
+ f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} "
+ f"with interval {interval * 1000} ms."
+ )
+ consumed_batches = 0
+
+ while consumed_batches < BATCHES_LIMIT:
+ try:
+ logger.debug("Polling for messages...")
+ polled_messages = await client.poll_messages(
+ stream=STREAM_NAME,
+ topic=TOPIC_NAME,
+ partition_id=PARTITION_ID,
+ polling_strategy=PollingStrategy.Next(),
+ count=MESSAGES_PER_BATCH,
+ auto_commit=True,
+ )
+ if not polled_messages:
+ logger.info("No messages found in current poll")
+ await asyncio.sleep(interval)
+ continue
+
+ for message in polled_messages:
+ handle_message(message)
+
+ consumed_batches += 1
+ logger.info(f"Consumed {len(polled_messages)} message(s).")
+ await asyncio.sleep(interval)
+ except Exception as error:
+ logger.exception(f"Exception occurred while consuming messages:
{error}")
+ break
+
+ logger.info(f"Consumed {consumed_batches} batches of messages, exiting.")
+
+
+def handle_message(message: ReceiveMessage):
+ payload = json.loads(message.payload().decode("utf-8"))
+ # `user_headers()` returns the explicitly typed `UserHeaders` mapping
+ # (a dict subclass) or None when the message carries no headers.
+ headers = message.user_headers()
+ message_type = get_message_type(headers)
+
+ logger.info(
+ f"Handling message at offset {message.offset()} "
+ f"with origin timestamp {message.origin_timestamp()}."
+ )
+ if headers is not None:
+ logger.info(f"Headers: {format_headers(headers)}")
+ # Opt into the convenient plain form.
+ logger.info(f"Plain headers: {format_headers(headers.to_plain())}")
+
+ if message_type == ORDER_CREATED_TYPE:
+ handle_order_created(payload)
+ elif message_type == ORDER_CONFIRMED_TYPE:
+ handle_order_confirmed(payload)
+ elif message_type == ORDER_REJECTED_TYPE:
+ handle_order_rejected(payload)
+ else:
+ logger.warning(f"Received unknown message type: {message_type}")
+
+
+def handle_order_created(order_created: Mapping[str, Any]):
+ logger.info(f"Order Created: {order_created}")
+
+
+def handle_order_confirmed(order_confirmed: Mapping[str, Any]):
+ logger.info(f"Order Confirmed: {order_confirmed}")
+
+
+def handle_order_rejected(order_rejected: Mapping[str, Any]):
+ logger.info(f"Order Rejected: {order_rejected}")
Review Comment:
Inline these. these do not add anything new to code.
##########
foreign/python/src/user_headers.rs:
##########
@@ -0,0 +1,857 @@
+// 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.
+
+use std::collections::BTreeMap;
+use std::collections::hash_map::DefaultHasher;
+use std::hash::{Hash, Hasher};
+
+use iggy::prelude::{
+ HeaderField, HeaderKey as RustHeaderKey, HeaderKind, HeaderValue as
RustHeaderValue,
+};
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+use pyo3::pyclass::CompareOp;
+use pyo3::types::{PyBool, PyBytes, PyDict, PyFloat, PyInt, PyString};
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_complex_enum,
gen_stub_pymethods};
+
+type RustUserHeaders = BTreeMap<RustHeaderKey, RustHeaderValue>;
+
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+/// Typed key for an Iggy user header.
+///
+/// Use these constructors when the header key must preserve an explicit
+/// wire type instead of using the common string-key dictionary form.
+pub enum HeaderKey {
+ /// Raw bytes key. The byte length must be 1..=255.
+ Raw { value: Py<PyBytes> },
+ /// UTF-8 string key. The encoded byte length must be 1..=255.
+ String { value: String },
+ /// Boolean key.
+ Bool { value: bool },
+ /// Signed 8-bit integer key.
+ Int8 { value: i8 },
+ /// Signed 16-bit integer key.
+ Int16 { value: i16 },
+ /// Signed 32-bit integer key.
+ Int32 { value: i32 },
+ /// Signed 64-bit integer key.
+ Int64 { value: i64 },
+ /// Signed 128-bit integer key.
+ Int128 { value: i128 },
+ /// Unsigned 8-bit integer key.
+ UnsignedInt8 { value: u8 },
+ /// Unsigned 16-bit integer key.
+ UnsignedInt16 { value: u16 },
+ /// Unsigned 32-bit integer key.
+ UnsignedInt32 { value: u32 },
+ /// Unsigned 64-bit integer key.
+ UnsignedInt64 { value: u64 },
+ /// Unsigned 128-bit integer key.
+ UnsignedInt128 { value: u128 },
+ /// 32-bit floating point key.
+ Float32 { value: f32 },
+ /// 64-bit floating point key.
+ Float64 { value: f64 },
+}
+
+#[gen_stub_pyclass_complex_enum]
+#[pyclass]
+/// Typed value for an Iggy user header.
+///
+/// Use these constructors when the header value must preserve an explicit
+/// wire type instead of using the common Python scalar dictionary form.
+pub enum HeaderValue {
+ /// Raw bytes value. The byte length must be 1..=255.
+ Raw { value: Py<PyBytes> },
+ /// UTF-8 string value. The encoded byte length must be 1..=255.
+ String { value: String },
+ /// Boolean value.
+ Bool { value: bool },
+ /// Signed 8-bit integer value.
+ Int8 { value: i8 },
+ /// Signed 16-bit integer value.
+ Int16 { value: i16 },
+ /// Signed 32-bit integer value.
+ Int32 { value: i32 },
+ /// Signed 64-bit integer value.
+ Int64 { value: i64 },
+ /// Signed 128-bit integer value.
+ Int128 { value: i128 },
+ /// Unsigned 8-bit integer value.
+ UnsignedInt8 { value: u8 },
+ /// Unsigned 16-bit integer value.
+ UnsignedInt16 { value: u16 },
+ /// Unsigned 32-bit integer value.
+ UnsignedInt32 { value: u32 },
+ /// Unsigned 64-bit integer value.
+ UnsignedInt64 { value: u64 },
+ /// Unsigned 128-bit integer value.
+ UnsignedInt128 { value: u128 },
+ /// 32-bit floating point value.
+ Float32 { value: f32 },
+ /// 64-bit floating point value.
+ Float64 { value: f64 },
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl HeaderKey {
+ pub fn __hash__(&self, py: Python<'_>) -> PyResult<isize> {
+ Ok(py_hash(header_identity_hash(self.identity(py)?)))
+ }
+
+ pub fn __richcmp__(
+ &self,
+ py: Python<'_>,
+ other: &Bound<'_, PyAny>,
+ op: CompareOp,
+ ) -> PyResult<Py<PyAny>> {
+ header_key_richcmp(py, self.identity(py)?, other, op)
+ }
+
+ pub fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
+ self.repr(py)
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl HeaderValue {
+ pub fn __hash__(&self, py: Python<'_>) -> PyResult<isize> {
+ Ok(py_hash(header_identity_hash(self.identity(py)?)))
+ }
+
+ pub fn __richcmp__(
+ &self,
+ py: Python<'_>,
+ other: &Bound<'_, PyAny>,
+ op: CompareOp,
+ ) -> PyResult<Py<PyAny>> {
+ header_value_richcmp(py, self.identity(py)?, other, op)
+ }
+
+ pub fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
+ self.repr(py)
+ }
+}
+
+impl HeaderKey {
+ fn to_rust(&self, py: Python<'_>) -> PyResult<RustHeaderKey> {
+ match self {
+ HeaderKey::Raw { value } => {
+
RustHeaderKey::try_from(value.extract::<Vec<u8>>(py)?).map_err(to_value_error)
+ }
+ HeaderKey::String { value } => {
+ RustHeaderKey::try_from(value.as_str()).map_err(to_value_error)
+ }
+ HeaderKey::Bool { value } => Ok((*value).into()),
+ HeaderKey::Int8 { value } => Ok((*value).into()),
+ HeaderKey::Int16 { value } => Ok((*value).into()),
+ HeaderKey::Int32 { value } => Ok((*value).into()),
+ HeaderKey::Int64 { value } => Ok((*value).into()),
+ HeaderKey::Int128 { value } => Ok((*value).into()),
+ HeaderKey::UnsignedInt8 { value } => Ok((*value).into()),
+ HeaderKey::UnsignedInt16 { value } => Ok((*value).into()),
+ HeaderKey::UnsignedInt32 { value } => Ok((*value).into()),
+ HeaderKey::UnsignedInt64 { value } => Ok((*value).into()),
+ HeaderKey::UnsignedInt128 { value } => Ok((*value).into()),
+ HeaderKey::Float32 { value } => checked_float32(*value),
+ HeaderKey::Float64 { value } => Ok((*value).into()),
+ }
+ }
+
+ fn from_rust<'a>(py: Python<'a>, value: &RustHeaderKey) ->
PyResult<Bound<'a, Self>> {
+ let key = match value.kind() {
+ HeaderKind::Raw => HeaderKey::Raw {
+ value: PyBytes::new(py,
value.as_raw().map_err(to_value_error)?).unbind(),
+ },
+ HeaderKind::String => HeaderKey::String {
+ value: value.as_str().map_err(to_value_error)?.to_string(),
+ },
+ HeaderKind::Bool => HeaderKey::Bool {
+ value: value.as_bool().map_err(to_value_error)?,
+ },
+ HeaderKind::Int8 => HeaderKey::Int8 {
+ value: value.as_int8().map_err(to_value_error)?,
+ },
+ HeaderKind::Int16 => HeaderKey::Int16 {
+ value: value.as_int16().map_err(to_value_error)?,
+ },
+ HeaderKind::Int32 => HeaderKey::Int32 {
+ value: value.as_int32().map_err(to_value_error)?,
+ },
+ HeaderKind::Int64 => HeaderKey::Int64 {
+ value: value.as_int64().map_err(to_value_error)?,
+ },
+ HeaderKind::Int128 => HeaderKey::Int128 {
+ value: value.as_int128().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint8 => HeaderKey::UnsignedInt8 {
+ value: value.as_uint8().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint16 => HeaderKey::UnsignedInt16 {
+ value: value.as_uint16().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint32 => HeaderKey::UnsignedInt32 {
+ value: value.as_uint32().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint64 => HeaderKey::UnsignedInt64 {
+ value: value.as_uint64().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint128 => HeaderKey::UnsignedInt128 {
+ value: value.as_uint128().map_err(to_value_error)?,
+ },
+ HeaderKind::Float32 => HeaderKey::Float32 {
+ value: value.as_float32().map_err(to_value_error)?,
+ },
+ HeaderKind::Float64 => HeaderKey::Float64 {
+ value: value.as_float64().map_err(to_value_error)?,
+ },
+ };
+ key.into_pyobject(py)
+ }
+
+ fn identity(&self, py: Python<'_>) -> PyResult<HeaderIdentity> {
+ match self {
+ HeaderKey::Raw { value } =>
Ok(HeaderIdentity::raw(value.extract::<Vec<u8>>(py)?)),
+ HeaderKey::String { value } =>
Ok(HeaderIdentity::string(value.as_bytes().to_vec())),
+ HeaderKey::Bool { value } => Ok(HeaderIdentity::bool(*value)),
+ HeaderKey::Int8 { value } => Ok(HeaderIdentity::int8(*value)),
+ HeaderKey::Int16 { value } => Ok(HeaderIdentity::int16(*value)),
+ HeaderKey::Int32 { value } => Ok(HeaderIdentity::int32(*value)),
+ HeaderKey::Int64 { value } => Ok(HeaderIdentity::int64(*value)),
+ HeaderKey::Int128 { value } => Ok(HeaderIdentity::int128(*value)),
+ HeaderKey::UnsignedInt8 { value } =>
Ok(HeaderIdentity::uint8(*value)),
+ HeaderKey::UnsignedInt16 { value } =>
Ok(HeaderIdentity::uint16(*value)),
+ HeaderKey::UnsignedInt32 { value } =>
Ok(HeaderIdentity::uint32(*value)),
+ HeaderKey::UnsignedInt64 { value } =>
Ok(HeaderIdentity::uint64(*value)),
+ HeaderKey::UnsignedInt128 { value } =>
Ok(HeaderIdentity::uint128(*value)),
+ HeaderKey::Float32 { value } =>
Ok(HeaderIdentity::float32(*value)),
+ HeaderKey::Float64 { value } =>
Ok(HeaderIdentity::float64(*value)),
+ }
+ }
+
+ fn repr(&self, py: Python<'_>) -> PyResult<String> {
+ match self {
+ HeaderKey::Raw { value } => Ok(format!(
+ "HeaderKey.Raw({})",
+ value.bind(py).repr()?.extract::<String>()?
+ )),
+ HeaderKey::String { value } =>
Ok(format!("HeaderKey.String({value:?})")),
+ HeaderKey::Bool { value } =>
Ok(format!("HeaderKey.Bool({value})")),
+ HeaderKey::Int8 { value } =>
Ok(format!("HeaderKey.Int8({value})")),
+ HeaderKey::Int16 { value } =>
Ok(format!("HeaderKey.Int16({value})")),
+ HeaderKey::Int32 { value } =>
Ok(format!("HeaderKey.Int32({value})")),
+ HeaderKey::Int64 { value } =>
Ok(format!("HeaderKey.Int64({value})")),
+ HeaderKey::Int128 { value } =>
Ok(format!("HeaderKey.Int128({value})")),
+ HeaderKey::UnsignedInt8 { value } =>
Ok(format!("HeaderKey.UnsignedInt8({value})")),
+ HeaderKey::UnsignedInt16 { value } =>
Ok(format!("HeaderKey.UnsignedInt16({value})")),
+ HeaderKey::UnsignedInt32 { value } =>
Ok(format!("HeaderKey.UnsignedInt32({value})")),
+ HeaderKey::UnsignedInt64 { value } =>
Ok(format!("HeaderKey.UnsignedInt64({value})")),
+ HeaderKey::UnsignedInt128 { value } =>
Ok(format!("HeaderKey.UnsignedInt128({value})")),
+ HeaderKey::Float32 { value } =>
Ok(format!("HeaderKey.Float32({value:?})")),
+ HeaderKey::Float64 { value } =>
Ok(format!("HeaderKey.Float64({value:?})")),
+ }
+ }
+}
+
+impl HeaderValue {
+ fn to_rust(&self, py: Python<'_>) -> PyResult<RustHeaderValue> {
+ match self {
+ HeaderValue::Raw { value } => {
+
RustHeaderValue::try_from(value.extract::<Vec<u8>>(py)?).map_err(to_value_error)
+ }
+ HeaderValue::String { value } => {
+
RustHeaderValue::try_from(value.as_str()).map_err(to_value_error)
+ }
+ HeaderValue::Bool { value } => Ok((*value).into()),
+ HeaderValue::Int8 { value } => Ok((*value).into()),
+ HeaderValue::Int16 { value } => Ok((*value).into()),
+ HeaderValue::Int32 { value } => Ok((*value).into()),
+ HeaderValue::Int64 { value } => Ok((*value).into()),
+ HeaderValue::Int128 { value } => Ok((*value).into()),
+ HeaderValue::UnsignedInt8 { value } => Ok((*value).into()),
+ HeaderValue::UnsignedInt16 { value } => Ok((*value).into()),
+ HeaderValue::UnsignedInt32 { value } => Ok((*value).into()),
+ HeaderValue::UnsignedInt64 { value } => Ok((*value).into()),
+ HeaderValue::UnsignedInt128 { value } => Ok((*value).into()),
+ HeaderValue::Float32 { value } => checked_float32(*value),
+ HeaderValue::Float64 { value } => Ok((*value).into()),
+ }
+ }
+
+ fn from_rust<'a>(py: Python<'a>, value: &RustHeaderValue) ->
PyResult<Bound<'a, Self>> {
+ let value = match value.kind() {
+ HeaderKind::Raw => HeaderValue::Raw {
+ value: PyBytes::new(py,
value.as_raw().map_err(to_value_error)?).unbind(),
+ },
+ HeaderKind::String => HeaderValue::String {
+ value: value.as_str().map_err(to_value_error)?.to_string(),
+ },
+ HeaderKind::Bool => HeaderValue::Bool {
+ value: value.as_bool().map_err(to_value_error)?,
+ },
+ HeaderKind::Int8 => HeaderValue::Int8 {
+ value: value.as_int8().map_err(to_value_error)?,
+ },
+ HeaderKind::Int16 => HeaderValue::Int16 {
+ value: value.as_int16().map_err(to_value_error)?,
+ },
+ HeaderKind::Int32 => HeaderValue::Int32 {
+ value: value.as_int32().map_err(to_value_error)?,
+ },
+ HeaderKind::Int64 => HeaderValue::Int64 {
+ value: value.as_int64().map_err(to_value_error)?,
+ },
+ HeaderKind::Int128 => HeaderValue::Int128 {
+ value: value.as_int128().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint8 => HeaderValue::UnsignedInt8 {
+ value: value.as_uint8().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint16 => HeaderValue::UnsignedInt16 {
+ value: value.as_uint16().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint32 => HeaderValue::UnsignedInt32 {
+ value: value.as_uint32().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint64 => HeaderValue::UnsignedInt64 {
+ value: value.as_uint64().map_err(to_value_error)?,
+ },
+ HeaderKind::Uint128 => HeaderValue::UnsignedInt128 {
+ value: value.as_uint128().map_err(to_value_error)?,
+ },
+ HeaderKind::Float32 => HeaderValue::Float32 {
+ value: value.as_float32().map_err(to_value_error)?,
+ },
+ HeaderKind::Float64 => HeaderValue::Float64 {
+ value: value.as_float64().map_err(to_value_error)?,
+ },
+ };
+ value.into_pyobject(py)
+ }
+
+ fn identity(&self, py: Python<'_>) -> PyResult<HeaderIdentity> {
+ match self {
+ HeaderValue::Raw { value } =>
Ok(HeaderIdentity::raw(value.extract::<Vec<u8>>(py)?)),
+ HeaderValue::String { value } =>
Ok(HeaderIdentity::string(value.as_bytes().to_vec())),
+ HeaderValue::Bool { value } => Ok(HeaderIdentity::bool(*value)),
+ HeaderValue::Int8 { value } => Ok(HeaderIdentity::int8(*value)),
+ HeaderValue::Int16 { value } => Ok(HeaderIdentity::int16(*value)),
+ HeaderValue::Int32 { value } => Ok(HeaderIdentity::int32(*value)),
+ HeaderValue::Int64 { value } => Ok(HeaderIdentity::int64(*value)),
+ HeaderValue::Int128 { value } =>
Ok(HeaderIdentity::int128(*value)),
+ HeaderValue::UnsignedInt8 { value } =>
Ok(HeaderIdentity::uint8(*value)),
+ HeaderValue::UnsignedInt16 { value } =>
Ok(HeaderIdentity::uint16(*value)),
+ HeaderValue::UnsignedInt32 { value } =>
Ok(HeaderIdentity::uint32(*value)),
+ HeaderValue::UnsignedInt64 { value } =>
Ok(HeaderIdentity::uint64(*value)),
+ HeaderValue::UnsignedInt128 { value } =>
Ok(HeaderIdentity::uint128(*value)),
+ HeaderValue::Float32 { value } =>
Ok(HeaderIdentity::float32(*value)),
+ HeaderValue::Float64 { value } =>
Ok(HeaderIdentity::float64(*value)),
+ }
+ }
+
+ fn repr(&self, py: Python<'_>) -> PyResult<String> {
+ match self {
+ HeaderValue::Raw { value } => Ok(format!(
+ "HeaderValue.Raw({})",
+ value.bind(py).repr()?.extract::<String>()?
+ )),
+ HeaderValue::String { value } =>
Ok(format!("HeaderValue.String({value:?})")),
+ HeaderValue::Bool { value } =>
Ok(format!("HeaderValue.Bool({value})")),
+ HeaderValue::Int8 { value } =>
Ok(format!("HeaderValue.Int8({value})")),
+ HeaderValue::Int16 { value } =>
Ok(format!("HeaderValue.Int16({value})")),
+ HeaderValue::Int32 { value } =>
Ok(format!("HeaderValue.Int32({value})")),
+ HeaderValue::Int64 { value } =>
Ok(format!("HeaderValue.Int64({value})")),
+ HeaderValue::Int128 { value } =>
Ok(format!("HeaderValue.Int128({value})")),
+ HeaderValue::UnsignedInt8 { value } =>
Ok(format!("HeaderValue.UnsignedInt8({value})")),
+ HeaderValue::UnsignedInt16 { value } => {
+ Ok(format!("HeaderValue.UnsignedInt16({value})"))
+ }
+ HeaderValue::UnsignedInt32 { value } => {
+ Ok(format!("HeaderValue.UnsignedInt32({value})"))
+ }
+ HeaderValue::UnsignedInt64 { value } => {
+ Ok(format!("HeaderValue.UnsignedInt64({value})"))
+ }
+ HeaderValue::UnsignedInt128 { value } => {
+ Ok(format!("HeaderValue.UnsignedInt128({value})"))
+ }
+ HeaderValue::Float32 { value } =>
Ok(format!("HeaderValue.Float32({value:?})")),
+ HeaderValue::Float64 { value } =>
Ok(format!("HeaderValue.Float64({value:?})")),
+ }
+ }
+}
+
+#[derive(PartialEq, Eq, Hash)]
+struct HeaderIdentity {
+ kind: u8,
+ value: Vec<u8>,
+}
+
+impl HeaderIdentity {
+ fn raw(value: Vec<u8>) -> Self {
+ Self {
+ kind: HeaderKind::Raw.as_code(),
+ value,
+ }
+ }
+
+ fn string(value: Vec<u8>) -> Self {
+ Self {
+ kind: HeaderKind::String.as_code(),
+ value,
+ }
+ }
+
+ fn bool(value: bool) -> Self {
+ Self {
+ kind: HeaderKind::Bool.as_code(),
+ value: vec![u8::from(value)],
+ }
+ }
+
+ fn int8(value: i8) -> Self {
+ Self {
+ kind: HeaderKind::Int8.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int16(value: i16) -> Self {
+ Self {
+ kind: HeaderKind::Int16.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int32(value: i32) -> Self {
+ Self {
+ kind: HeaderKind::Int32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int64(value: i64) -> Self {
+ Self {
+ kind: HeaderKind::Int64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn int128(value: i128) -> Self {
+ Self {
+ kind: HeaderKind::Int128.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint8(value: u8) -> Self {
+ Self {
+ kind: HeaderKind::Uint8.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint16(value: u16) -> Self {
+ Self {
+ kind: HeaderKind::Uint16.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint32(value: u32) -> Self {
+ Self {
+ kind: HeaderKind::Uint32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint64(value: u64) -> Self {
+ Self {
+ kind: HeaderKind::Uint64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn uint128(value: u128) -> Self {
+ Self {
+ kind: HeaderKind::Uint128.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn float32(value: f32) -> Self {
+ Self {
+ kind: HeaderKind::Float32.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+
+ fn float64(value: f64) -> Self {
+ Self {
+ kind: HeaderKind::Float64.as_code(),
+ value: value.to_le_bytes().to_vec(),
+ }
+ }
+}
+
+pub(crate) fn py_user_headers_to_rust(
+ py: Python<'_>,
+ headers: &Bound<'_, PyDict>,
+) -> PyResult<RustUserHeaders> {
+ // Each key/value pair is converted independently: typed `HeaderKey` /
+ // `HeaderValue` are used as-is, while plain Python scalars are converted
on
+ // a best-effort basis.
+ let mut rust_headers = BTreeMap::new();
+ for (key, value) in headers.iter() {
+ let key = py_header_key_to_rust(py, &key)?;
+ let value = py_header_value_to_rust(py, &value)?;
+ rust_headers.insert(key, value);
+ }
+ Ok(rust_headers)
+}
+
+pub(crate) fn rust_user_headers_to_py<'a>(
+ py: Python<'a>,
+ headers: RustUserHeaders,
+) -> PyResult<Bound<'a, UserHeaders>> {
+ // Always expose the explicitly typed dict[HeaderKey, HeaderValue] so that
+ // no wire-type information is silently dropped. Callers who prefer the
+ // convenient plain form opt in through `UserHeaders.to_plain`.
+ let result = Bound::new(py, UserHeaders)?;
+ let mapping = result.as_any();
+ for (key, value) in headers {
+ let key = HeaderKey::from_rust(py, &key)?;
+ let value = HeaderValue::from_rust(py, &value)?;
+ mapping.set_item(key, value)?;
+ }
+ Ok(result)
+}
+
+/// User headers dictionary returned by `ReceiveMessage.user_headers`.
+///
+/// This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping
+/// operations work) that additionally exposes `to_plain` for the convenient
+/// scalar form.
+#[gen_stub_pyclass]
+#[pyclass(extends=PyDict)]
+pub struct UserHeaders;
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl UserHeaders {
+ /// Wraps a mapping so its entries gain the `to_plain` helper.
+ ///
+ /// Accepts a dict whose keys and values can each independently be
+ /// `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool |
+ /// int | float`). The inherited `dict` initializer copies the provided
+ /// mapping.
+ #[new]
+ #[pyo3(signature = (mapping=None))]
+ pub fn new(
+ #[gen_stub(override_type(type_repr = "dict[typing.Any, typing.Any] |
None"))]
+ mapping: Option<&Bound<'_, PyAny>>,
+ ) -> Self {
+ let _ = mapping;
+ UserHeaders
+ }
+
+ /// Converts these headers into the convenient plain dictionary form.
+ ///
+ /// Every header kind maps losslessly onto a Python scalar, so this never
+ /// loses information; it only returns an error if a stored field cannot be
+ /// decoded.
+ #[gen_stub(override_return_type(
+ type_repr = "dict[str | bytes | bool | int | float, str | bytes | bool
| int | float]"
+ ))]
+ pub fn to_plain<'a>(slf: &Bound<'a, Self>) -> PyResult<Bound<'a, PyDict>> {
+ let py = slf.py();
+ let dict = slf.as_any().cast::<PyDict>()?;
+ let headers = py_user_headers_to_rust(py, dict)?;
+ rust_user_headers_to_plain_py(py, headers)
Review Comment:
Let's rename to `to_scalar_dict`.
##########
examples/python/message-headers/producer.py:
##########
@@ -0,0 +1,237 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+import secrets
+import time
+from collections.abc import Mapping
+from typing import NamedTuple
+
+from apache_iggy import HeaderKey, HeaderValue, IggyClient, StreamDetails,
TopicDetails
+from apache_iggy import SendMessage as Message
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
+
+PlainHeaderValue = str | bytes | bool | int | float
+PlainHeaders = dict[str, PlainHeaderValue]
+TypedHeaders = dict[HeaderKey, HeaderValue]
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+class SerializedMessage(NamedTuple):
+ message_type: str
+ payload: str
+ headers: PlainHeaders | TypedHeaders
+
+
+class MessagesGenerator:
+ def __init__(self):
+ self.order_id = 0
+
+ def generate(self) -> SerializedMessage:
Review Comment:
Let's use an iterator here instead of a full blown class
##########
examples/python/message-headers/producer.py:
##########
@@ -0,0 +1,237 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+import secrets
+import time
+from collections.abc import Mapping
+from typing import NamedTuple
+
+from apache_iggy import HeaderKey, HeaderValue, IggyClient, StreamDetails,
TopicDetails
+from apache_iggy import SendMessage as Message
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
+
+PlainHeaderValue = str | bytes | bool | int | float
+PlainHeaders = dict[str, PlainHeaderValue]
+TypedHeaders = dict[HeaderKey, HeaderValue]
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+class SerializedMessage(NamedTuple):
+ message_type: str
+ payload: str
+ headers: PlainHeaders | TypedHeaders
+
+
+class MessagesGenerator:
+ def __init__(self):
+ self.order_id = 0
+
+ def generate(self) -> SerializedMessage:
+ self.order_id += 1
+ message_type = self.order_id % 3
+
+ if message_type == 0:
+ payload = {
+ "orderId": f"order-{self.order_id}",
+ "customerId": f"customer-{secrets.randbelow(100)}",
+ "amount": secrets.randbelow(10000) + 1,
+ }
+ return self._serialize(ORDER_CREATED_TYPE, payload)
+
+ if message_type == 1:
+ payload = {
+ "orderId": f"order-{self.order_id // 3}",
+ "timestamp": int(time.time() * 1000),
+ }
+ return self._serialize(ORDER_CONFIRMED_TYPE, payload)
+
+ payload = {
+ "orderId": f"order-{self.order_id // 3}",
+ "reason": "Insufficient balance",
+ }
+ return self._serialize(ORDER_REJECTED_TYPE, payload)
+
+ def _serialize(
+ self, message_type: str, payload: Mapping[str, object]
+ ) -> SerializedMessage:
+ encoded_payload = json.dumps(payload)
+ if self.order_id % 5 == 0:
+ # Add typed headers similar to the Rust core
+ typed_headers: TypedHeaders = {
+ HeaderKey.String("message-type"):
HeaderValue.String(message_type),
+ HeaderKey.String("content-type"): HeaderValue.String(
+ "application/json"
+ ),
+ HeaderKey.String("schema-version"):
HeaderValue.UnsignedInt16(1),
+ HeaderKey.String("created-at-ms"): HeaderValue.UnsignedInt64(
+ int(time.time() * 1000)
+ ),
+ HeaderKey.String("retryable"): HeaderValue.Bool(
+ message_type == ORDER_REJECTED_TYPE
+ ),
+ HeaderKey.String("priority-score"): HeaderValue.Float32(
+ (self.order_id % 100) / 100
+ ),
+ HeaderKey.String("trace-bin"): HeaderValue.Raw(
+ f"trace-{self.order_id}".encode()
+ ),
+ HeaderKey.UnsignedInt32(self.order_id):
HeaderValue.String("order-id"),
+ }
+ return SerializedMessage(message_type, encoded_payload,
typed_headers)
+
+ # Add easy-to-use `dict[str, str | bytes | bool | int | float]` headers
+ # which will be translated into typed headers by the Python SDK
+ plain_headers: PlainHeaders = {
+ "message-type": message_type,
+ "content-type": "application/json",
+ "schema-version": 1,
+ "created-at-ms": int(time.time() * 1000),
+ "retryable": message_type == ORDER_REJECTED_TYPE,
+ "priority-score": (self.order_id % 100) / 100,
+ "trace-bin": f"trace-{self.order_id}".encode(),
+ }
+ return SerializedMessage(message_type, encoded_payload, plain_headers)
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy client, e.g. "
+ "'iggy+tcp://iggy:[email protected]:8090'"
+ ),
+ default="iggy+tcp://iggy:[email protected]:8090",
+ nargs="?",
+ type=str,
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def main():
+ args: ArgNamespace = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+ logger.info("Connecting to Iggy")
+ await client.connect()
+ logger.info("Connected")
+ await init_system(client)
+ await produce_messages(client)
+
+
+async def init_system(client: IggyClient):
Review Comment:
Current system catches stream creation and topic creation independently,
logs errors, and continues with an invalid system. Unify the try/catches and
exit if stream/topic is not created
##########
examples/python/message-headers/consumer.py:
##########
@@ -0,0 +1,170 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any, NamedTuple
+
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+ UserHeaders,
+)
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy client, e.g. "
+ "'iggy+tcp://iggy:[email protected]:8090'"
+ ),
+ default="iggy+tcp://iggy:[email protected]:8090",
+ nargs="?",
+ type=str,
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def main():
+ args: ArgNamespace = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+ logger.info("Connecting to Iggy")
+ await client.connect()
+ logger.info("Connected")
+ await consume_messages(client)
+
+
+async def consume_messages(client: IggyClient):
+ interval = 0.5
+ logger.info(
+ f"Messages will be consumed from stream: {STREAM_NAME}, "
+ f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} "
+ f"with interval {interval * 1000} ms."
+ )
+ consumed_batches = 0
+
+ while consumed_batches < BATCHES_LIMIT:
+ try:
+ logger.debug("Polling for messages...")
+ polled_messages = await client.poll_messages(
+ stream=STREAM_NAME,
+ topic=TOPIC_NAME,
+ partition_id=PARTITION_ID,
+ polling_strategy=PollingStrategy.Next(),
+ count=MESSAGES_PER_BATCH,
+ auto_commit=True,
+ )
+ if not polled_messages:
+ logger.info("No messages found in current poll")
+ await asyncio.sleep(interval)
+ continue
+
+ for message in polled_messages:
+ handle_message(message)
+
+ consumed_batches += 1
+ logger.info(f"Consumed {len(polled_messages)} message(s).")
+ await asyncio.sleep(interval)
+ except Exception as error:
+ logger.exception(f"Exception occurred while consuming messages:
{error}")
+ break
+
+ logger.info(f"Consumed {consumed_batches} batches of messages, exiting.")
+
+
+def handle_message(message: ReceiveMessage):
+ payload = json.loads(message.payload().decode("utf-8"))
+ # `user_headers()` returns the explicitly typed `UserHeaders` mapping
+ # (a dict subclass) or None when the message carries no headers.
+ headers = message.user_headers()
+ message_type = get_message_type(headers)
+
+ logger.info(
+ f"Handling message at offset {message.offset()} "
+ f"with origin timestamp {message.origin_timestamp()}."
+ )
+ if headers is not None:
+ logger.info(f"Headers: {format_headers(headers)}")
+ # Opt into the convenient plain form.
+ logger.info(f"Plain headers: {format_headers(headers.to_plain())}")
+
+ if message_type == ORDER_CREATED_TYPE:
+ handle_order_created(payload)
+ elif message_type == ORDER_CONFIRMED_TYPE:
+ handle_order_confirmed(payload)
+ elif message_type == ORDER_REJECTED_TYPE:
+ handle_order_rejected(payload)
+ else:
+ logger.warning(f"Received unknown message type: {message_type}")
Review Comment:
match/case can be utilized instead of if/else
##########
examples/python/message-headers/producer.py:
##########
@@ -0,0 +1,237 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+import secrets
+import time
+from collections.abc import Mapping
+from typing import NamedTuple
+
+from apache_iggy import HeaderKey, HeaderValue, IggyClient, StreamDetails,
TopicDetails
+from apache_iggy import SendMessage as Message
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
+
+PlainHeaderValue = str | bytes | bool | int | float
+PlainHeaders = dict[str, PlainHeaderValue]
+TypedHeaders = dict[HeaderKey, HeaderValue]
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+class SerializedMessage(NamedTuple):
Review Comment:
Let's use a dataclass instead here: ` @dataclass(frozen=True, slots=True)`
##########
foreign/python/tests/test_message_operations.py:
##########
Review Comment:
Let's add the following test cases as well:
- send message with duplicate headers should error
- a message that has > 100k bytes is not accepted, so this should be tested
and should fail.
-
##########
examples/python/message-headers/consumer.py:
##########
@@ -0,0 +1,170 @@
+# 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.
+
+import argparse
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any, NamedTuple
+
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+ UserHeaders,
+)
+from loguru import logger
+
+STREAM_NAME = "message-headers-stream"
+TOPIC_NAME = "orders"
+PARTITION_ID = 0
+BATCHES_LIMIT = 5
+MESSAGES_PER_BATCH = 10
+
+ORDER_CREATED_TYPE = "OrderCreated"
+ORDER_CONFIRMED_TYPE = "OrderConfirmed"
+ORDER_REJECTED_TYPE = "OrderRejected"
Review Comment:
You can utilize `StrEnum` here instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
