Make DeliveryObservableProxy.EnumerateFeed async

Author: winklertomas

Kontent.Ai.Delivery.Rx.Tests/Kontent.Ai.Delivery.Rx.Tests.csproj

@@ -47,6 +47,7 @@
     <PackageReference Include="FakeItEasy" Version="8.0.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
     <PackageReference Include="RichardSzalay.MockHttp" Version="6.0.0" />
+    <PackageReference Include="System.Reactive" Version="6.0.1" />
     <PackageReference Include="xunit" Version="2.9.2" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
       <PrivateAssets>all</PrivateAssets>

Kontent.Ai.Delivery.Rx/DeliveryItemsFeedExtensions.cs

@@ -0,0 +1,59 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive.Linq;
+using Kontent.Ai.Delivery.Abstractions;
+
+namespace Kontent.Ai.Delivery.Rx;
+
+/// <summary>
+/// Provides extension methods for working with <see cref="IDeliveryItemsFeed{T}"/> instances in a reactive programming context.
+/// </summary>
+public static class DeliveryItemsFeedExtensions
+{
+    /// <summary>
+    /// Converts an <see cref="IDeliveryItemsFeed{T}"/> into an <see cref="IObservable{T}"/> sequence.
+    /// </summary>
+    /// <typeparam name="T">The type of content items in the feed.</typeparam>
+    /// <param name="feed">The <see cref="IDeliveryItemsFeed{T}"/> instance to convert.</param>
+    /// <returns>
+    /// An <see cref="IObservable{T}"/> sequence that emits items retrieved from the feed.
+    /// If the feed is <c>null</c>, an empty observable sequence is returned.
+    /// </returns>
+    /// <exception cref="Exception">
+    /// Propagates any exceptions that occur during the retrieval of items from the feed.
+    /// </exception>
+    public static IObservable<T> ToObservable<T>(this IDeliveryItemsFeed<T> feed) where T : class
+    {
+        if (feed == null)
+        {
+            return Observable.Empty<T>();
+        }
+        return Observable.Create<T>(async observer =>
+        {
+            try
+            {
+                await foreach (var item in EnumerateFeed(feed))
+                {
+                    observer.OnNext(item);
+                }
+                observer.OnCompleted();
+            }
+            catch (Exception ex)
+            {
+                observer.OnError(ex);
+            }
+        });
+    }
+
+    private static async IAsyncEnumerable<T> EnumerateFeed<T>(IDeliveryItemsFeed<T> feed) where T : class
+    {
+        while (feed.HasMoreResults)
+        {
+            var batch = await feed.FetchNextBatchAsync();
+            foreach (var contentItem in batch.Items)
+            {
+                yield return contentItem;
+            }
+        }
+    }
+}

Kontent.Ai.Delivery.Rx/DeliveryObservableProxy.cs

@@ -2,6 +2,8 @@
 using System.Collections.Generic;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
+using System.Reactive;
+using System.Threading.Tasks;
 using Kontent.Ai.Delivery.Abstractions;
 
 namespace Kontent.Ai.Delivery.Rx
@@ -104,8 +106,7 @@ public IObservable<T> GetItemsFeedObservable<T>(params IQueryParameter[] paramet
         /// <returns>The <see cref="IObservable{T}"/> that represents the content items. If no query parameters are specified, all content items are returned.</returns>
         public IObservable<T> GetItemsFeedObservable<T>(IEnumerable<IQueryParameter> parameters) where T : class
         {
-            var feed = DeliveryClient?.GetItemsFeed<T>(parameters);
-            return feed == null ? null : EnumerateFeed(feed)?.ToObservable();
+            return DeliveryClient?.GetItemsFeed<T>(parameters).ToObservable();
         }
 
         /// <summary>
@@ -127,8 +128,7 @@ public IObservable<IUsedInItem> GetItemUsedInObservable(string codename, params
         /// <returns>The <see cref="IObservable{IUsedInItem}"/> that represents the parent content items for the specified content item. If no query parameters are specified, parents in default language are returned.</returns>
         public IObservable<IUsedInItem> GetItemUsedInObservable(string codename, IEnumerable<IQueryParameter> parameters)
         {
-            var feed = DeliveryClient?.GetItemUsedIn(codename, parameters);
-            return feed == null ? null : EnumerateFeed(feed)?.ToObservable();
+            return DeliveryClient?.GetItemUsedIn(codename, parameters).ToObservable();
         }
 
         /// <summary>
@@ -150,8 +150,7 @@ public IObservable<IUsedInItem> GetAssetUsedInObservable(string codename, params
         /// <returns>The <see cref="IObservable{IUsedInItem}"/> that represents the parent content items for the specified content item. If no query parameters are specified, parents in default language are returned.</returns>
         public IObservable<IUsedInItem> GetAssetUsedInObservable(string codename, IEnumerable<IQueryParameter> parameters)
         {
-            var feed = DeliveryClient?.GetAssetUsedIn(codename, parameters);
-            return feed == null ? null : EnumerateFeed(feed)?.ToObservable();
+            return DeliveryClient?.GetAssetUsedIn(codename, parameters).ToObservable();
         }
 
         /// <summary>
@@ -246,17 +245,6 @@ public IObservable<ILanguage> GetLanguagesObservable(params IQueryParameter[] pa
 
         #endregion
         #region "Private methods"
-        private static IEnumerable<T> EnumerateFeed<T>(IDeliveryItemsFeed<T> feed) where T : class
-        {
-            while (feed.HasMoreResults)
-            {
-                foreach (var contentItem in feed.FetchNextBatchAsync().Result.Items)
-                {
-                    yield return contentItem;
-                }
-            }
-        }
-
         private static IObservable<T> GetObservableOfOne<T>(Func<T> responseFactory)
         {
             return Observable.Create((IObserver<T> observer) =>

Kontent.Ai.Delivery.Rx/Kontent.Ai.Delivery.Rx.csproj

@@ -25,10 +25,10 @@
 
   <ItemGroup>
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
-    <PackageReference Include="System.Reactive" Version="5.0.0" />
+    <PackageReference Include="System.Reactive" Version="6.0.1" />
     <PackageReference Include="Microsoft.Sourcelink.Github" Version="1.1.1" PrivateAssets="All" />
-    <None Include="../README.md" Pack="true" PackagePath=""/>
-    <None Include="../kai-logo-symbol-color-rgb.png" Pack="true" PackagePath=""/>
+    <None Include="../README.md" Pack="true" PackagePath="" />
+    <None Include="../kai-logo-symbol-color-rgb.png" Pack="true" PackagePath="" />
   </ItemGroup>
 
   <ItemGroup>

Kontent.Ai.Delivery/DeliveryClient.cs

@@ -372,6 +372,11 @@ public async Task<IDeliverySyncResponse> GetSyncAsync(string continuationToken)
         /// <returns>The <see cref="DeliveryUsedInItems"/> instance that can be used to enumerate through content item parents for the specified item codename. If no query parameters are specified, default language parents are enumerated.</returns>
         public IDeliveryItemsFeed<IUsedInItem> GetItemUsedIn(string codename, IEnumerable<IQueryParameter> parameters = null)
         {
+            if (codename == null)
+            {
+                throw new ArgumentNullException(nameof(codename), "The codename of a content item is not specified.");
+            }
+
             ValidateUsedInParameters(parameters);
             var endpointUrl = UrlBuilder.GetItemUsedInUrl(codename, parameters);
 
@@ -386,6 +391,11 @@ public IDeliveryItemsFeed<IUsedInItem> GetItemUsedIn(string codename, IEnumerabl
         /// <returns>The <see cref="DeliveryUsedInItems"/> instance that can be used to enumerate through asset parents for the specified asset codename. If no query parameters are specified, default language parents are enumerated.</returns>
         public IDeliveryItemsFeed<IUsedInItem> GetAssetUsedIn(string codename, IEnumerable<IQueryParameter> parameters = null)
         {
+            if (codename == null)
+            {
+                throw new ArgumentNullException(nameof(codename), "The codename of an asset is not specified.");
+            }
+
             ValidateUsedInParameters(parameters);
             var endpointUrl = UrlBuilder.GetAssetUsedInUrl(codename, parameters);