Implement grid and directory discovery (#71)

crates/libremetaverse-types/src/compat.rs

@@ -1003,7 +1003,153 @@ pub struct Span<T>(pub PhantomData<fn(T)>);
 
 pub struct RandomSource;
 
-pub struct IAsyncEnumerable<T>(pub PhantomData<fn() -> T>);
+struct AsyncEnumerableState<T> {
+    values: std::collections::VecDeque<T>,
+    closed: bool,
+    waiters: Vec<Waker>,
+    cancellation: Option<CancellationToken>,
+}
+
+/// A cloneable, producer-driven equivalent of `IAsyncEnumerable<T>`.
+///
+/// Native managers keep a producer clone and publish correlated protocol
+/// results. Consumers call [`IAsyncEnumerable::next`] until it returns `None`.
+pub struct IAsyncEnumerable<T>(Arc<Mutex<AsyncEnumerableState<T>>>);
+
+impl<T> Clone for IAsyncEnumerable<T> {
+    fn clone(&self) -> Self {
+        Self(Arc::clone(&self.0))
+    }
+}
+
+impl<T> Default for IAsyncEnumerable<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<T> IAsyncEnumerable<T> {
+    #[must_use]
+    pub fn new() -> Self {
+        Self(Arc::new(Mutex::new(AsyncEnumerableState {
+            values: std::collections::VecDeque::new(),
+            closed: false,
+            waiters: Vec::new(),
+            cancellation: None,
+        })))
+    }
+
+    #[must_use]
+    pub fn with_cancellation(cancellation: CancellationToken) -> Self {
+        let value = Self::new();
+        value
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .cancellation = Some(cancellation);
+        value
+    }
+
+    /// Publishes one item and wakes pending consumers.
+    pub fn push(&self, value: T) {
+        let waiters = {
+            let mut state = self
+                .0
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            if state
+                .cancellation
+                .as_ref()
+                .is_some_and(CancellationToken::is_cancellation_requested)
+            {
+                state.closed = true;
+            }
+            if state.closed {
+                return;
+            }
+            state.values.push_back(value);
+            std::mem::take(&mut state.waiters)
+        };
+        for waiter in waiters {
+            waiter.wake();
+        }
+    }
+
+    /// Completes the sequence after already queued items are consumed.
+    pub fn close(&self) {
+        let waiters = {
+            let mut state = self
+                .0
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            state.closed = true;
+            std::mem::take(&mut state.waiters)
+        };
+        for waiter in waiters {
+            waiter.wake();
+        }
+    }
+
+    #[must_use]
+    pub fn next(&self) -> AsyncEnumerableNext<T> {
+        let cancellation = self
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .cancellation
+            .clone()
+            .map(|token| token.cancelled());
+        AsyncEnumerableNext {
+            state: Arc::clone(&self.0),
+            registered: false,
+            cancellation,
+        }
+    }
+}
+
+pub struct AsyncEnumerableNext<T> {
+    state: Arc<Mutex<AsyncEnumerableState<T>>>,
+    registered: bool,
+    cancellation: Option<CancellationFuture>,
+}
+
+impl<T> Future for AsyncEnumerableNext<T> {
+    type Output = Option<T>;
+
+    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+        let should_register = !self.registered;
+        if let Some(cancelled) = self.cancellation.as_mut()
+            && Pin::new(cancelled).poll(context).is_ready()
+        {
+            self.state
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .closed = true;
+        }
+        let outcome = {
+            let mut state = self
+                .state
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            if let Some(value) = state.values.pop_front() {
+                Some(Poll::Ready(Some(value)))
+            } else if state.closed {
+                Some(Poll::Ready(None))
+            } else {
+                if should_register {
+                    state.waiters.push(context.waker().clone());
+                }
+                None
+            }
+        };
+        if let Some(outcome) = outcome {
+            outcome
+        } else {
+            self.registered = true;
+            Poll::Pending
+        }
+    }
+}
 
 pub struct ImmutableDictionary<TKey, TValue>(pub PhantomData<fn(TKey, TValue)>);