diff --git a/Duplicati/Library/Backend/S3/Duplicati.Library.Backend.S3.csproj b/Duplicati/Library/Backend/S3/Duplicati.Library.Backend.S3.csproj index 45404ac18..cd9d50882 100644 --- a/Duplicati/Library/Backend/S3/Duplicati.Library.Backend.S3.csproj +++ b/Duplicati/Library/Backend/S3/Duplicati.Library.Backend.S3.csproj @@ -13,7 +13,7 @@ - + diff --git a/Duplicati/Library/Backend/S3/S3MinioClient.cs b/Duplicati/Library/Backend/S3/S3MinioClient.cs index a4d5b63fb..39e5bd99d 100644 --- a/Duplicati/Library/Backend/S3/S3MinioClient.cs +++ b/Duplicati/Library/Backend/S3/S3MinioClient.cs @@ -19,7 +19,6 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. using System.Runtime.CompilerServices; -using System.Threading.Channels; using Duplicati.Library.Common.IO; using Duplicati.Library.Interface; using Duplicati.Library.Utility; @@ -56,27 +55,6 @@ namespace Duplicati.Library.Backend m_dnsHost = servername; } - private static async IAsyncEnumerable ToAsyncEnumerable( - IObservable observable, - [EnumeratorCancellation] CancellationToken cancellationToken = default) - { - var channel = Channel.CreateUnbounded(); // Buffered channel for async iteration - - using var subscription = observable.Subscribe( - item => channel.Writer.TryWrite(item), - ex => channel.Writer.TryComplete(ex), - () => channel.Writer.TryComplete() - ); - - while (await channel.Reader.WaitToReadAsync(cancellationToken)) - { - while (channel.Reader.TryRead(out var item)) - { - yield return item; - } - } - } - public async IAsyncEnumerable ListBucketAsync(string bucketName, string prefix, bool recursive, [EnumeratorCancellation] CancellationToken cancellationToken) { await ThrowExceptionIfBucketDoesNotExist(bucketName, cancellationToken).ConfigureAwait(false); @@ -84,11 +62,7 @@ namespace Duplicati.Library.Backend if (!string.IsNullOrWhiteSpace(prefix)) prefix = Util.AppendDirSeparator(prefix, "/"); - var observable = await Utility.Utility.WithTimeout(m_timeouts.ListTimeout, cancellationToken, ct - => m_client.ListObjectsAsync(new ListObjectsArgs().WithBucket(bucketName).WithPrefix(prefix).WithRecursive(recursive), ct) - ).ConfigureAwait(false); - - await foreach (var obj in ToAsyncEnumerable(observable, cancellationToken).ConfigureAwait(false)) + await foreach (var obj in Utility.Utility.WithPerItemTimeout(m_client.ListObjectsEnumAsync(new ListObjectsArgs().WithBucket(bucketName).WithPrefix(prefix).WithRecursive(recursive), cancellationToken), m_timeouts.ListTimeout, cancellationToken)) { if (obj.Key == prefix || !obj.Key.StartsWith(prefix)) continue; diff --git a/Duplicati/Library/Utility/Utility.cs b/Duplicati/Library/Utility/Utility.cs index e1a56a9fc..ba620206f 100644 --- a/Duplicati/Library/Utility/Utility.cs +++ b/Duplicati/Library/Utility/Utility.cs @@ -29,6 +29,7 @@ using System.Linq; using System.Net; using System.Net.NetworkInformation; using System.Net.Sockets; +using System.Runtime.CompilerServices; using System.Runtime.Versioning; using System.Security.Cryptography; using System.Security.Cryptography.X509Certificates; @@ -1766,6 +1767,47 @@ namespace Duplicati.Library.Utility } } + /// + /// Wraps an async enumerable in a timeout observing enumerable + /// + /// The type of the items in the enumerable + /// The source enumerable + /// The timeout to observe for each item + /// The cancellation token for the outer operation + /// The wrapped enumerable + public static async IAsyncEnumerable WithPerItemTimeout( + IAsyncEnumerable source, + TimeSpan timeoutPerItem, + [EnumeratorCancellation] CancellationToken outerToken) + { + using var timeoutPolicy = new CancellationTokenSource(); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(timeoutPolicy.Token, outerToken); + + await using var enumerator = await WithTimeout(timeoutPerItem, outerToken, _ => source.GetAsyncEnumerator(linked.Token)).ConfigureAwait(false); + while (true) + { + timeoutPolicy.CancelAfter(timeoutPerItem); + Task moveNextTask; + try + { + moveNextTask = enumerator.MoveNextAsync().AsTask(); + var completed = await Task.WhenAny(moveNextTask, Task.Delay(Timeout.Infinite, timeoutPolicy.Token)); + if (completed != moveNextTask || timeoutPolicy.IsCancellationRequested) + throw new TimeoutException($"Timeout while waiting for next item ({timeoutPerItem.TotalSeconds}s)"); + + if (!moveNextTask.Result) + break; + + yield return enumerator.Current; + } + finally + { + timeoutPolicy.Token.ThrowIfCancellationRequested(); + timeoutPolicy.CancelAfter(Timeout.Infinite); // Reset for next item + } + } + } + /// /// Wraps the stream in a timeout observing stream ///