FlowsContainer.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Cleipnir.ResilientFunctions;
using Cleipnir.ResilientFunctions.Storage;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

namespace Cleipnir.Flows;

public class FlowsContainer : IDisposable
{
    internal readonly IServiceProvider ServiceProvider;
    internal readonly FunctionsRegistry FunctionRegistry;
    private readonly Dictionary<string, Type> _registeredFlows = new();
    private readonly Lock _lock = new();

    public FunctionsRegistry Functions => FunctionRegistry;

    public FlowsContainer(IFunctionStore flowStore, IServiceProvider serviceProvider, Options options)
    {
        ServiceProvider = serviceProvider;

        if (options.UnhandledExceptionHandler == null && serviceProvider.GetService<ILogger>() != null)
        {
            var logger = serviceProvider.GetRequiredService<ILogger>();
            options = new Options(
                    unhandledExceptionHandler: ex => logger.LogError(ex, "Unhandled exception in Cleipnir"),
                    retentionPeriod: options.RetentionPeriod,
                    retentionCleanUpFrequency: options.RetentionCleanUpFrequency,
                    enableWatchdogs: options.EnableWatchdogs,
                    watchdogCheckFrequency: options.WatchdogCheckFrequency,
                    messagesPullFrequency: options.MessagesPullFrequency,
                    messagesDefaultMaxWaitForCompletion: options.MessagesDefaultMaxWaitForCompletion,
                    delayStartup: options.DelayStartup,
                    maxParallelRetryInvocations: options.MaxParallelRetryInvocations,
                    serializer: options.Serializer,
                    utcNow: options.UtcNow,
                    replicaHeartbeatFrequency: options.ReplicaHeartbeatFrequency
                );
        }

        FunctionRegistry = new FunctionsRegistry(flowStore, options.MapToSettings());
    }

    internal void EnsureNoExistingRegistration(string flowName, Type flowType)
    {
        lock (_lock)
            if (_registeredFlows.TryGetValue(flowName, out var existingFlowType) && flowType != existingFlowType)
                throw new InvalidOperationException($"Flow with name '{flowName}' for type '{flowType}' has already been registered for different type: '{existingFlowType}'");
            else
                _registeredFlows[flowName] = flowType;
    }

    public void Dispose() => FunctionRegistry.Dispose();
    public Task ShutdownGracefully(TimeSpan? maxWait = null) => FunctionRegistry.ShutdownGracefully(maxWait);

    public Flows<TFlow> RegisterAnonymousFlow<TFlow>(Func<TFlow>? flowFactory = null, string? flowName = null, FlowOptions? options = null) where TFlow : Flow
    {
        flowName ??= typeof(TFlow).Name;
        return new Flows<TFlow>(flowName, flowsContainer: this, options ?? new FlowOptions(), flowFactory);
    }

    public Flows<TFlow, TParam> RegisterAnonymousFlow<TFlow, TParam>(Func<TFlow>? flowFactory = null, string? flowName = null, FlowOptions? options = null) where TFlow : Flow<TParam> where TParam : notnull
    {
        flowName ??= typeof(TFlow).Name;
        return new Flows<TFlow, TParam>(flowName, flowsContainer: this, options ?? new FlowOptions(), flowFactory);
    }

    public Flows<TFlow, TParam, TResult> RegisterAnonymousFlow<TFlow, TParam, TResult>(Func<TFlow>? flowFactory = null, string? flowName = null, FlowOptions? options = null) where TFlow : Flow<TParam, TResult> where TResult : notnull where TParam : notnull
    {
        flowName ??= typeof(TFlow).Name;
        return new Flows<TFlow, TParam, TResult>(flowName, flowsContainer: this, options ?? new FlowOptions(), flowFactory);
    }

    public static FlowsContainer Create(
        IServiceProvider? serviceProvider = null,
        IFunctionStore? functionStore = null,
        Options? options = null)
        => new(
            functionStore ?? new InMemoryFunctionStore(),
            serviceProvider ?? new ServiceCollection().BuildServiceProvider(),
            options ?? Options.Default
        );
}