Add StreamFailureHandler

src/Orleans/AmqpProtocol/Configuration/ConfigureQueueOptions.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Escendit Ltd. All Rights Reserved.
+// Licensed under the MIT. See LICENSE.txt file in the solution root for full license information.
+
+namespace Escendit.Orleans.Streaming.RabbitMQ.AmqpProtocol.Configuration;
+
+using global::Orleans.Streams;
+using Handlers;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+/// <inheritdoc />
+internal class ConfigureQueueOptions : IConfigureOptions<QueueOptions>
+{
+    private readonly ILoggerFactory _loggerFactory;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ConfigureQueueOptions"/> class.
+    /// </summary>
+    /// <param name="loggerFactory">The logger.</param>
+    public ConfigureQueueOptions(ILoggerFactory loggerFactory)
+    {
+        _loggerFactory = loggerFactory;
+    }
+
+    /// <inheritdoc />
+    public void Configure(QueueOptions options)
+    {
+        ArgumentNullException.ThrowIfNull(options);
+        options.StreamFailureHandler = (queueId) =>
+            Task.FromResult<IStreamFailureHandler>(
+                new RabbitMqStreamDeliveryFailureHandler(_loggerFactory.CreateLogger<RabbitMqStreamDeliveryFailureHandler>(), queueId, options.ShouldFaultSubscriptionOnError));
+    }
+}

src/Orleans/AmqpProtocol/Handlers/RabbitMqStreamDeliveryFailureHandler.cs

@@ -0,0 +1,69 @@
+// Copyright (c) Escendit Ltd. All Rights Reserved.
+// Licensed under the MIT. See LICENSE.txt file in the solution root for full license information.
+
+namespace Escendit.Orleans.Streaming.RabbitMQ.AmqpProtocol.Handlers;
+
+using global::Orleans.Runtime;
+using global::Orleans.Streams;
+using Microsoft.Extensions.Logging;
+
+/// <summary>
+/// RabbitMQ Stream Failure Handler.
+/// </summary>
+internal partial class RabbitMqStreamDeliveryFailureHandler : IStreamFailureHandler
+{
+    private readonly ILogger _logger;
+    private readonly QueueId _queueId;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="RabbitMqStreamDeliveryFailureHandler"/> class.
+    /// </summary>
+    /// <param name="logger">The logger.</param>
+    /// <param name="queueId">The queue id.</param>
+    /// <param name="shouldFaultSubscriptionOnError">The value if it should fault on subscription error.</param>
+    public RabbitMqStreamDeliveryFailureHandler(ILogger<RabbitMqStreamDeliveryFailureHandler> logger, QueueId queueId, bool shouldFaultSubscriptionOnError = false)
+    {
+        _logger = logger;
+        _queueId = queueId;
+        ShouldFaultSubsriptionOnError = shouldFaultSubscriptionOnError;
+    }
+
+    /// <inheritdoc/>
+    public bool ShouldFaultSubsriptionOnError { get; }
+
+    /// <inheritdoc />
+    public Task OnDeliveryFailure(
+        GuidId subscriptionId,
+        string streamProviderName,
+        StreamId streamIdentity,
+        StreamSequenceToken sequenceToken)
+    {
+        LogDeliveryFailure(_queueId, subscriptionId.Guid, streamProviderName, streamIdentity, sequenceToken, ShouldFaultSubsriptionOnError);
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc/>
+    public Task OnSubscriptionFailure(
+        GuidId subscriptionId,
+        string streamProviderName,
+        StreamId streamIdentity,
+        StreamSequenceToken sequenceToken)
+    {
+        LogSubscriptionFailure(_queueId, subscriptionId.Guid, streamProviderName, streamIdentity, sequenceToken, ShouldFaultSubsriptionOnError);
+        return Task.CompletedTask;
+    }
+
+    [LoggerMessage(
+        EventId = 1000,
+        EventName = "Delivery Failure",
+        Level = LogLevel.Warning,
+        Message = "Delivery Failure with QueueId: {queueId}, Subscription: {subscriptionId}, Stream Provider: {streamProviderName}, Stream Id: {streamId}, Sequence Token: {streamSequenceToken} with {fault}")]
+    private partial void LogDeliveryFailure(QueueId queueId, Guid subscriptionId, string streamProviderName, StreamId streamId, StreamSequenceToken streamSequenceToken, bool fault);
+
+    [LoggerMessage(
+        EventId = 1010,
+        EventName = "Subscription Failure",
+        Level = LogLevel.Warning,
+        Message = "Subscription Failure with QueueId: {queueId}, Subscription: {subscriptionId}, Stream Provider: {streamProviderName}, Stream Id: {streamId}, Sequence Token: {streamSequenceToken} with {fault}")]
+    private partial void LogSubscriptionFailure(QueueId queueId, Guid subscriptionId, string streamProviderName, StreamId streamId, StreamSequenceToken streamSequenceToken, bool fault);
+}

src/Orleans/AmqpProtocol/Hosting/ClientBuilderExtensions.cs

@@ -53,4 +53,18 @@ public static IRabbitMqClientOptionsBuilder ConfigureStreamPubSub(this IRabbitMq
         clientBuilder.Configurator.ConfigureStreamPubSub(streamPubSubType);
         return clientBuilder;
     }
+
+    /// <summary>
+    /// Configure Default Stream Failure Handler.
+    /// </summary>
+    /// <param name="clientBuilder">The initial client options builder.</param>
+    /// <returns>The updated client options builder.</returns>
+    public static IRabbitMqClientOptionsBuilder ConfigureDefaultStreamDeliveryFailureHandler(this IRabbitMqClientOptionsBuilder clientBuilder)
+    {
+        clientBuilder.ConfigureServices(services =>
+        {
+            services.ConfigureOptions<ConfigureQueueOptions>();
+        });
+        return clientBuilder;
+    }
 }

src/Orleans/AmqpProtocol/Hosting/SiloBuilderExtensions.cs

@@ -54,4 +54,18 @@ public static IRabbitMqSiloOptionsBuilder ConfigureStreamPubSub(this IRabbitMqSi
         siloBuilder.Configurator.ConfigureStreamPubSub(streamPubSubType);
         return siloBuilder;
     }
+
+    /// <summary>
+    /// Configure Default Stream Failure Handler.
+    /// </summary>
+    /// <param name="siloOptionsBuilder">The initial silo options builder.</param>
+    /// <returns>The updated silo options builder.</returns>
+    public static IRabbitMqSiloOptionsBuilder ConfigureDefaultStreamDeliveryFailureHandler(this IRabbitMqSiloOptionsBuilder siloOptionsBuilder)
+    {
+        siloOptionsBuilder.ConfigureServices(services =>
+        {
+            services.ConfigureOptions<ConfigureQueueOptions>();
+        });
+        return siloOptionsBuilder;
+    }
 }

src/Orleans/RabbitMQ/Configuration/OptionsBase.cs

@@ -9,13 +9,29 @@ namespace Escendit.Orleans.Streaming.RabbitMQ.Configuration;
 /// <summary>
 /// Options Base.
 /// </summary>
-public abstract class OptionsBase
+public class OptionsBase
 {
     /// <summary>
-    /// Gets the stream failure handler.
+    /// Initializes a new instance of the <see cref="OptionsBase"/> class.
+    /// </summary>
+    protected OptionsBase()
+    {
+    }
+
+    /// <summary>
+    /// Gets or sets the stream failure handler.
     /// </summary>
     /// <value>The stream failure handler.</value>
     [Browsable(false)]
-    public Func<QueueId, Task<IStreamFailureHandler>> StreamFailureHandler { get; internal set; } = _ =>
+    [EditorBrowsable(EditorBrowsableState.Advanced)]
+    public Func<QueueId, Task<IStreamFailureHandler>> StreamFailureHandler { get; set; } = _ =>
         Task.FromResult<IStreamFailureHandler>(new NoOpStreamDeliveryFailureHandler());
+
+    /// <summary>
+    /// Gets or sets a value indicating whether it should fault on subscription error.
+    /// </summary>
+    /// <value>The flag if subscription should fault.</value>
+    [Browsable(false)]
+    [EditorBrowsable(EditorBrowsableState.Advanced)]
+    public bool ShouldFaultSubscriptionOnError { get; set; }
 }