hubcio commented on code in PR #3463: URL: https://github.com/apache/iggy/pull/3463#discussion_r3413290383
########## foreign/php/src/message_iterator.rs: ########## @@ -0,0 +1,99 @@ +// 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::sync::Arc; + +use ext_php_rs::{exception::PhpResult, php_class, php_impl, zend::ce}; +use futures::StreamExt; +use iggy::prelude::IggyConsumer as RustIggyConsumer; +use tokio::sync::Mutex; + +use crate::error::to_php_exception; +use crate::receive_message::ReceiveMessage; +use crate::runtime::runtime; + +#[php_class] +#[php(name = "Iggy\\MessageIterator")] +#[php(implements(ce = ce::iterator, stub = "\\Iterator"))] +pub struct MessageIterator { + pub(crate) inner: Arc<Mutex<RustIggyConsumer>>, + current: Option<ReceiveMessage>, + key: u64, + started: bool, +} + +impl MessageIterator { + pub(crate) fn new(inner: Arc<Mutex<RustIggyConsumer>>) -> Self { + Self { + inner, + current: None, + key: 0, + started: false, + } + } + + fn poll_next(&mut self) -> PhpResult { + let inner = self.inner.clone(); + + self.current = runtime().block_on(async move { + let mut inner = inner.lock().await; + + match inner.next().await { + Some(Ok(message)) => Ok(Some(ReceiveMessage { + inner: message.message, + partition_id: message.partition_id, + })), + Some(Err(err)) => Err(to_php_exception(err)), + None => Ok(None), + } + })?; + + Ok(()) + } +} + +#[php_impl] +impl MessageIterator { + pub fn current(&self) -> Option<ReceiveMessage> { + self.current.clone() + } + + pub fn key(&self) -> u64 { + self.key + } + + pub fn next(&mut self) -> PhpResult { + if self.current.is_some() { + self.key += 1; + } + + self.poll_next() + } + + pub fn rewind(&mut self) -> PhpResult { + if !self.started { + self.started = true; + self.poll_next()?; + } + + Ok(()) + } + + pub fn valid(&self) -> bool { Review Comment: worth a one-line note here on why the iterator is open-ended. the underlying sdk consumer stream never returns `None` for a live consumer - on an empty batch it re-arms the poll and parks, so `valid()` stays true and the `None => Ok(None)` arm in `poll_next` is dead during normal operation. net effect is a plain `foreach` over `iterMessages()` blocks forever unless the caller `break`s. readme covers the break, but the reason it never ends isn't obvious from this file. ########## foreign/php/src/message_iterator.rs: ########## @@ -0,0 +1,99 @@ +// 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::sync::Arc; + +use ext_php_rs::{exception::PhpResult, php_class, php_impl, zend::ce}; +use futures::StreamExt; +use iggy::prelude::IggyConsumer as RustIggyConsumer; +use tokio::sync::Mutex; + +use crate::error::to_php_exception; +use crate::receive_message::ReceiveMessage; +use crate::runtime::runtime; + +#[php_class] +#[php(name = "Iggy\\MessageIterator")] +#[php(implements(ce = ce::iterator, stub = "\\Iterator"))] +pub struct MessageIterator { + pub(crate) inner: Arc<Mutex<RustIggyConsumer>>, + current: Option<ReceiveMessage>, + key: u64, + started: bool, +} + +impl MessageIterator { + pub(crate) fn new(inner: Arc<Mutex<RustIggyConsumer>>) -> Self { + Self { + inner, + current: None, + key: 0, + started: false, + } + } + + fn poll_next(&mut self) -> PhpResult { Review Comment: this async block is identical to `next_message()` over in `consumer.rs` - same `inner.lock().await`, `inner.next().await`, and match arms building `ReceiveMessage`. worth pulling out a small free fn like `poll_one(inner: &Arc<Mutex<RustIggyConsumer>>) -> PhpResult<Option<ReceiveMessage>>` and calling it from both: here `self.current = poll_one(&self.inner)?`, there just `return poll_one(...)`. removes the only real duplication, about 14 lines. ########## foreign/php/src/message_iterator.rs: ########## @@ -0,0 +1,99 @@ +// 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::sync::Arc; + +use ext_php_rs::{exception::PhpResult, php_class, php_impl, zend::ce}; +use futures::StreamExt; +use iggy::prelude::IggyConsumer as RustIggyConsumer; +use tokio::sync::Mutex; + +use crate::error::to_php_exception; +use crate::receive_message::ReceiveMessage; +use crate::runtime::runtime; + +#[php_class] +#[php(name = "Iggy\\MessageIterator")] +#[php(implements(ce = ce::iterator, stub = "\\Iterator"))] +pub struct MessageIterator { + pub(crate) inner: Arc<Mutex<RustIggyConsumer>>, + current: Option<ReceiveMessage>, + key: u64, + started: bool, +} + +impl MessageIterator { + pub(crate) fn new(inner: Arc<Mutex<RustIggyConsumer>>) -> Self { + Self { + inner, + current: None, + key: 0, + started: false, + } + } + + fn poll_next(&mut self) -> PhpResult { + let inner = self.inner.clone(); + + self.current = runtime().block_on(async move { + let mut inner = inner.lock().await; + + match inner.next().await { + Some(Ok(message)) => Ok(Some(ReceiveMessage { + inner: message.message, + partition_id: message.partition_id, + })), + Some(Err(err)) => Err(to_php_exception(err)), + None => Ok(None), + } + })?; + + Ok(()) + } +} + +#[php_impl] +impl MessageIterator { + pub fn current(&self) -> Option<ReceiveMessage> { + self.current.clone() + } + + pub fn key(&self) -> u64 { + self.key + } + + pub fn next(&mut self) -> PhpResult { + if self.current.is_some() { + self.key += 1; + } + + self.poll_next() + } + + pub fn rewind(&mut self) -> PhpResult { Review Comment: no-op rewind after the first call is the right choice for a forward-only server cursor - you can't seek back, and re-polling here would skip the buffered `current`. only side effect is that re-`foreach`ing the same iterator object resumes instead of restarting. fine as-is, maybe a short doc line that the iterator is single-pass so nobody expects a second loop to replay from the start. ########## foreign/php/README.md: ########## @@ -190,6 +199,8 @@ iggy+tcp://iggy:[email protected]:8090?tls=true&tls_domain=localhost&tls_ca_file=/p polled. Pass an explicit partition id before the first poll. - `consumeMessages()` requires an explicit finite limit. It does not run forever by default. +- `iterMessages()` returns a PHP `Iterator` and can be used with `foreach`. Review Comment: the `AutoCommit::when()` caveat just below applies to `iterMessages()` too, not only the callback path. with `AutoCommit::interval`/`when` an offset can be stored as the message is polled, before the `foreach` body runs or breaks, so the same at-least-once gap exists. worth folding iterMessages into that note so it's documented for both paths. ########## foreign/php/src/consumer.rs: ########## @@ -128,6 +127,11 @@ impl IggyConsumer { Ok(consumed) } + + /// Returns an iterator over messages for use with foreach. + pub fn iter_messages(&self) -> MessageIterator { Review Comment: `iter_messages()` clones the same `Arc<Mutex<RustIggyConsumer>>` the callback path uses, so the iterator isn't an independent cursor. two iterators, or a `foreach` plus a `consumeMessages()` call on the same consumer, split the stream between them - whoever locks first drains the shared buffer and advances the offset. not unsafe, but a php dev will likely expect `iterMessages()` to be its own cursor, so worth a doc note on the shared-consumer semantics. -- 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]
