Example raft cluster consensus about dotnext HOT 45 CLOSED

using DotNext.Net.Cluster;
using DotNext.Net.Cluster.Consensus.Raft;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;

namespace Services.Replication
{
    internal sealed class ClusterConfigurator : IClusterMemberLifetime
    {
        internal static void LeaderChanged(ICluster cluster, IClusterMember? leader)
        {
            Debug.Assert(cluster is IRaftCluster);
            var term = ((IRaftCluster)cluster).Term;
            var timeout = ((IRaftCluster)cluster).ElectionTimeout;
            Console.WriteLine(leader is null
                ? "Consensus cannot be reached"
                : $"New cluster leader is elected. Leader address is {leader.EndPoint}");
            Console.WriteLine($"Term of local cluster member is {term}. Election timeout {timeout}");
        }

        public void Initialize(IRaftCluster cluster, IDictionary<string, string> metadata)
        {
            cluster.LeaderChanged += LeaderChanged;
        }

        public void Shutdown(IRaftCluster cluster)
        {
            cluster.LeaderChanged -= LeaderChanged;
        }
    }
}

 app.UseConsensusProtocolHandler()
                .RedirectToLeader(LeaderResource)
                .UseRouting()
                .UseEndpoints(endpoints =>
                {
                    endpoints.MapGet(LeaderResource, RedirectToLeader);
                });

 public class Startup
    {
        private readonly IConfiguration configuration;

        public Startup(IConfiguration configuration)
        {
            this.configuration = configuration;
            Newtonsoft.Json.JsonConvert.DefaultSettings = () => new CustomJsonSettings();
        }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            // Configuration of NewtonsoftJson
            services.AddControllers()
                .AddNewtonsoftJson();
            services.AddMvc()
                .SetCompatibilityVersion(CompatibilityVersion.Latest)
                .AddJsonOptions(options =>
                    options.JsonSerializerOptions.MaxDepth = 10)
                .AddNewtonsoftJson(options => 
                    options.SerializerSettings.MaxDepth = 10)
                .AddJsonOptions(options =>
                    options.JsonSerializerOptions.MaxDepth = 10);
            Newtonsoft.Json.JsonConvert.DefaultSettings = () => 
                new Newtonsoft.Json.JsonSerializerSettings() {
                    MaxDepth = 10
                };


            // Configuration of Cluster
            services.ConfigureCluster<ClusterConfigurator>()
               .AddSingleton<IHttpMessageHandlerFactory, RaftClientHandlerFactory>()
               .AddOptions()
               .AddRouting();

            var path = configuration[SimplePersistentState.LogLocation];
            if (!string.IsNullOrWhiteSpace(path))
            {
                services.AddSingleton<AppEventSource>();
                services.UsePersistenceEngine<IValueProvider, SimplePersistentState>()
                    .AddSingleton<IHostedService, DataModifier>();
            }
        }



        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            // app.UseHttpsRedirection();

            app.UseRouting();
            app.UseAuthorization();
            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });


            // Configuration of Cluster
            const string LeaderResource = "/leader";


            // Check
            app.UseConsensusProtocolHandler()
                .RedirectToLeader(LeaderResource)
                .UseRouting()
                .UseEndpoints(endpoints =>
                {
                    endpoints.MapGet(LeaderResource, RedirectToLeader);
                });
        }

        private static Task RedirectToLeader(HttpContext context)
        {
            var cluster = context.RequestServices.GetRequiredService<IRaftCluster>();
            return context.Response.WriteAsync($"Leader address is {cluster.Leader?.EndPoint}. Current address is {context.Connection.LocalIpAddress}:{context.Connection.LocalPort}", context.RequestAborted);
        }
    }

    public class Program
    {
        public static void Main(string[] args)
        {
            var config = new ConfigurationBuilder()
                .AddJsonFile("appsettings.json", optional: false)
                .Build();

            _ = RaftManager.Main(config);
            CreateHostBuilder(args).Build().Run();
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureWebHostDefaults(webBuilder =>
                {
                    webBuilder.UseStartup<Startup>();
                });
    }

public static class RaftManager
    {
        //private RaftCluster cluster = null;
        
        //public RaftManager(string[][] clusterPeers, int localPort)
        //{

        //    RaftCluster.NodeConfiguration config = 
        //        new RaftCluster.TcpConfiguration(new IPEndPoint(IPAddress.Loopback, localPort));

        //    // Cluster configuration
        //    config.Partitioning = true;


        //    for (int i = 0; i < clusterPeers.Length; i++)
        //    {
        //        config.Members.Add(new IPEndPoint(
        //            System.Net.IPAddress.Parse(clusterPeers[i][0]),
        //            Int32.Parse(clusterPeers[i][1])));
        //    }

        //    cluster = new RaftCluster(config);

        //}

        private static Task UseAspNetCoreHost(int port, string? persistentStorage = null)
        {
            var configuration = new Dictionary<string, string>
            {
                {"partitioning", "false"},
                {"lowerElectionTimeout", "150" },
                {"upperElectionTimeout", "300" },
                {"members:0", "https://localhost:63755"},
                {"requestJournal:memoryLimit", "5" },
                {"requestJournal:expiration", "00:01:00" }
            };
            if (!string.IsNullOrEmpty(persistentStorage))
                configuration[SimplePersistentState.LogLocation] = persistentStorage;
            return new HostBuilder().ConfigureWebHost(webHost =>
            {
                webHost.UseKestrel(options =>
                {
                    options.ListenLocalhost(port, listener => listener.UseHttps(LoadCertificate()));
                })
                .UseStartup<Startup>();
            })
            .ConfigureLogging(builder => builder.AddConsole().SetMinimumLevel(LogLevel.Error))
            .ConfigureAppConfiguration(builder => builder.AddInMemoryCollection(configuration))
            .JoinCluster()
            .Build()
            .RunAsync();
        }

        private static async Task UseConfiguration(RaftCluster.NodeConfiguration config, string? persistentStorage)
        {
            config.Members.Add(new IPEndPoint(IPAddress.Loopback, 63755));
            var loggerFactory = new LoggerFactory();
            var loggerOptions = new ConsoleLoggerOptions
            {
                LogToStandardErrorThreshold = LogLevel.Warning
            };
            loggerFactory.AddProvider(new ConsoleLoggerProvider(new FakeOptionsMonitor<ConsoleLoggerOptions>(loggerOptions)));
            config.LoggerFactory = loggerFactory;

            using var cluster = new RaftCluster(config);
            cluster.LeaderChanged += ClusterConfigurator.LeaderChanged;
            var modifier = default(DataModifier?);
            if (!string.IsNullOrEmpty(persistentStorage))
            {
                var state = new SimplePersistentState(persistentStorage, new AppEventSource());
                cluster.AuditTrail = state;
                modifier = new DataModifier(cluster, state);
            }
            await cluster.StartAsync(CancellationToken.None);
            await (modifier?.StartAsync(CancellationToken.None) ?? Task.CompletedTask);
            using var handler = new CancelKeyPressHandler();
            Console.CancelKeyPress += handler.Handler;
            await handler.WaitAsync();
            Console.CancelKeyPress -= handler.Handler;
            await (modifier?.StopAsync(CancellationToken.None) ?? Task.CompletedTask);
            await cluster.StopAsync(CancellationToken.None);
        }

        private static Task UseUdpTransport(int port, string? persistentStorage)
        {
            var configuration = new RaftCluster.UdpConfiguration(new IPEndPoint(IPAddress.Loopback, port))
            {
                LowerElectionTimeout = 150,
                UpperElectionTimeout = 300,
                DatagramSize = 1024
            };
            return UseConfiguration(configuration, persistentStorage);
        }

        private static Task UseTcpTransport(int port, string? persistentStorage, bool useSsl)
        {
            var configuration = new RaftCluster.TcpConfiguration(new IPEndPoint(IPAddress.Loopback, port))
            {
                LowerElectionTimeout = 150,
                UpperElectionTimeout = 300,
                TransmissionBlockSize = 4096,
                SslOptions = useSsl ? CreateSslOptions() : null
            };

            return UseConfiguration(configuration, persistentStorage);

            static SslOptions CreateSslOptions()
            {
                var options = new SslOptions();
                options.ServerOptions.ServerCertificate = LoadCertificate();
                options.ClientOptions.TargetHost = "localhost";
                options.ClientOptions.RemoteCertificateValidationCallback = RaftClientHandlerFactory.AllowCertificate;
                return options;
            }
        }

        private static Task StartNode(string protocol, int port, string? persistentStorage = null)
        {
            switch (protocol.ToLowerInvariant())
            {
                case "http":
                case "https":
                    return UseAspNetCoreHost(port, persistentStorage);
                case "udp":
                    return UseUdpTransport(port, persistentStorage);
                case "tcp":
                    return UseTcpTransport(port, persistentStorage, false);
                case "tcp+ssl":
                    return UseTcpTransport(port, persistentStorage, true);
                default:
                    Console.Error.WriteLine("Unsupported protocol type");
                    Environment.ExitCode = 1;
                    return Task.CompletedTask;
            }
        }

        public static async Task Main(IConfiguration configuration)
        {
            string protocol = configuration["ReplicationConfig:Protocol"];
            int port = int.Parse(configuration["ReplicationConfig:LocalPort"]);
            string persistentStorage = (configuration["ReplicationConfig:PersistentStorage"] == "") ? null : configuration["ReplicationConfig:PersistentStorage"];

            switch (persistentStorage)
            {
                case null:
                    await StartNode(protocol, port);
                    break;
                default:
                    await StartNode(protocol, port, persistentStorage);
                    break;
            }
        }

        private static X509Certificate2 LoadCertificate()
        {
            var certificate = new X509Certificate2("C:\\Users\\User\\Documents\\Symbiotic\\Git\\ToP-PMS\\PaymentManagementServer\\PaymentManagementServer.Services\\Certificates\\node.pfx",
                "1234", 
                X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
            return certificate;
        }
    }

    internal sealed class Int64LogEntry : BinaryTransferObject<long>, IRaftLogEntry
    {
        internal Int64LogEntry()
        {
            Timestamp = DateTimeOffset.UtcNow;
        }

        bool ILogEntry.IsSnapshot => false;

        public long Term { get; set; }

        public DateTimeOffset Timestamp { get; }
    }
}

    public class ObjectLogEntry : BinaryTransferObject<Object>, IRaftLogEntry
    {
        internal ObjectLogEntry()
    {
        Timestamp = DateTimeOffset.UtcNow;
    }

    bool ILogEntry.IsSnapshot => false;

    public Object Term { get; set; }

    public DateTimeOffset Timestamp { get; }
}

struct SubtractCommand
{
	public int X { get; set; }
	public int Y { get; set; }
}

 public class MyInterpreter : CommandInterpreter
	{
		public static SubtractCommand stateSubtract;
		public static NegateCommand stateNegate;

		[CommandHandler]
		public async ValueTask SubtractAsync(SubtractCommand command, CancellationToken token)
		{
			stateSubtract = command;

			Debug.WriteLine("I received Subtract {0}", command);
		}

		[CommandHandler(IsSnapshotHandler = true)]
		public async ValueTask HandleSnapshotSubtractAsync(SubtractCommand command, CancellationToken token)
		{
		}

		[CommandHandler]
		public async ValueTask NegateAsync(NegateCommand command, CancellationToken token)
		{
			stateNegate = command;

			Debug.WriteLine("I received Negate {0}", command);
		}

		[CommandHandler(IsSnapshotHandler = true)]
		public async ValueTask HandleSnapshotNegateAsync(NegateCommand command, CancellationToken token)
		{
		}
	}

sealed class CommandFormatter : IFormatter<SubtractCommand>, IFormatter<NegateCommand>
	{
		public static readonly CommandFormatter Instance = new CommandFormatter();

		private CommandFormatter()
		{
		}

		async ValueTask IFormatter<SubtractCommand>.SerializeAsync<TWriter>(SubtractCommand command, TWriter writer, CancellationToken token)
		{
			await writer.WriteInt32Async(command.X, true, token);
			await writer.WriteInt32Async(command.Y, true, token);
		}

		async ValueTask<SubtractCommand> IFormatter<SubtractCommand>.DeserializeAsync<TReader>(TReader reader, CancellationToken token)
		{
			return new SubtractCommand
			{
				X = await reader.ReadInt32Async(true, token),
				Y = await reader.ReadInt32Async(true, token)
			};
		}

		async ValueTask IFormatter<NegateCommand>.SerializeAsync<TWriter>(NegateCommand command, TWriter writer, CancellationToken token)
		{
			await writer.WriteInt32Async(command.X, true, token);
		}

		async ValueTask<NegateCommand> IFormatter<NegateCommand>.DeserializeAsync<TReader>(TReader reader, CancellationToken token)
		{
			return new NegateCommand
			{
				X = await reader.ReadInt32Async(true, token),
			};
		}
	}

	[CommandAttribute(1, Formatter = typeof(CommandFormatter), FormatterMember = nameof(CommandFormatter.Instance))]
	public struct NegateCommand
	{
		public int X { get; set; }
	}

	[CommandAttribute(0, Formatter = typeof(CommandFormatter), FormatterMember = nameof(CommandFormatter.Instance))]
	public struct SubtractCommand
	{
		public int X { get; set; }
		public int Y { get; set; }
	}

    [CommandAttribute(2, Formatter = typeof(CommandFormatter), FormatterMember = nameof(CommandFormatter.Instance))]
    public struct DataCommand
    {
        public string StringData { get; set; }

        public int Test { get; set; }
    }

        [Theory]
        [MemberData(nameof(GetDataForStringEncoding))]
        public static async Task WriteReadStringAsync(IAsyncBinaryReaderWriterSource source, Encoding encoding, LengthFormat? lengthFormat)
        {
            await using (source)
            {
                const string value = "Hello, world!&*(@&*(fghjwgfwffgw Привет, мир!";
                var writer = source.CreateWriter();
                await writer.WriteAsync(value.AsMemory(), encoding, lengthFormat);

                var reader = source.CreateReader();
                var result = await (lengthFormat is null ?
                    reader.ReadStringAsync(encoding.GetByteCount(value), encoding) :
                    reader.ReadStringAsync(lengthFormat.GetValueOrDefault(), encoding));
                Equal(value, result);
            }
        }

 var result = await (lengthFormat is null ?
                    reader.ReadStringAsync(encoding.GetByteCount(value), encoding) :
                    reader.ReadStringAsync(lengthFormat.GetValueOrDefault(), encoding));