Added tests for JSON log entries

Author: sakno

src/DotNext.Tests/Net/Cluster/Consensus/Raft/StateMachine/JsonStateMachine.cs

@@ -0,0 +1,32 @@
+using System.Diagnostics.CodeAnalysis;
+using DotNext.IO;
+
+namespace DotNext.Net.Cluster.Consensus.Raft.StateMachine;
+
+using Text.Json;
+
+[Experimental("DOTNEXT001")]
+internal sealed class JsonStateMachine : SimpleStateMachine
+{
+    private readonly List<TestJsonObject> entries = new();
+    
+    public JsonStateMachine(DirectoryInfo location)
+        : base(new(Path.Combine(location.FullName, "db")))
+    {
+    }
+
+    protected override ValueTask RestoreAsync(FileInfo snapshotFile, CancellationToken token)
+        => ValueTask.CompletedTask;
+
+    protected override ValueTask PersistAsync(IAsyncBinaryWriter writer, CancellationToken token)
+        => ValueTask.CompletedTask;
+
+    protected override async ValueTask<bool> ApplyAsync(LogEntry entry, CancellationToken token)
+    {
+        var content = await JsonSerializable<TestJsonObject>.TransformAsync(entry, token);
+        entries.Add(content);
+        return false;
+    }
+    
+    internal IReadOnlyList<TestJsonObject> Entries => entries;
+}

src/DotNext.Tests/Net/Cluster/Consensus/Raft/StateMachine/WriteAheadLogTests.cs

@@ -8,6 +8,7 @@
 namespace DotNext.Net.Cluster.Consensus.Raft.StateMachine;
 
 using Buffers.Binary;
+using Text.Json;
 using static IO.DataTransferObject;
 using LogEntryConsumer = IO.Log.LogEntryConsumer<IRaftLogEntry, Missing>;
 using LogEntryList = IO.Log.LogEntryProducer<IRaftLogEntry>;
@@ -348,4 +349,24 @@ public static void EmptyReader()
         var reader = new WriteAheadLog.LogEntryReader();
         Empty(reader);
     }
+    
+    [Fact]
+    public static async Task JsonSerialization()
+    {
+        var dir = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
+        await using var stateMachine = new JsonStateMachine(new(dir));
+        await using var wal = new WriteAheadLog(new() { Location = dir }, stateMachine);
+        
+        await wal.AppendJsonAsync(new TestJsonObject { StringField = "Entry1" });
+        var index = await wal.AppendJsonAsync(new TestJsonObject { StringField = "Entry2" });
+        await wal.CommitAsync(index);
+        await wal.WaitForApplyAsync(index);
+        Equal(2, stateMachine.Entries.Count);
+
+        var payload = stateMachine.Entries[0];
+        Equal("Entry1", payload.StringField.Value);
+
+        payload = stateMachine.Entries[1];
+        Equal("Entry2", payload.StringField.Value);
+    }
 }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentStateExtensions.cs

@@ -2,6 +2,7 @@
 
 using Buffers.Binary;
 using IO.Log;
+using Text.Json;
 
 /// <summary>
 /// Provides various extension methods for <see cref="IPersistentState"/> interface.
@@ -62,4 +63,18 @@ public static ValueTask<long> AppendAsync<T>(this IPersistentState state, T payl
         CancellationToken token = default)
         where T : IBinaryFormattable<T>
         => state.AppendAsync<BinaryLogEntry<T>>(new() { Content = payload, Term = state.Term, Context = context }, token);
+
+    /// <summary>
+    /// Appends JSON objec to the log tail.
+    /// </summary>
+    /// <param name="state">The log.</param>
+    /// <param name="payload">The log entry payload.</param>
+    /// <param name="context">The optional context to be passed to the state machine.</param>
+    /// <param name="token">The token that can be used to cancel the operation.</param>
+    /// <typeparam name="T">The type of the binary object.</typeparam>
+    /// <returns>The index of the added command within the log.</returns>
+    public static ValueTask<long> AppendJsonAsync<T>(this IPersistentState state, T payload, object? context = null,
+        CancellationToken token = default)
+        where T : IJsonSerializable<T>
+        => state.AppendAsync<JsonLogEntry<T>>(new() { Content = payload, Term = state.Term, Context = context }, token);
 }

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftClusterExtensions.cs

@@ -1,12 +1,15 @@
 namespace DotNext.Net.Cluster.Consensus.Raft;
 
+using Buffers.Binary;
+using Text.Json;
+
 /// <summary>
 /// Represents extension methods for <see cref="IRaftCluster"/> interface.
 /// </summary>
 public static class RaftClusterExtensions
 {
     /// <summary>
-    /// Appends a new log entry and ensures that it is replicated and committed.
+    /// Appends binary log entry and ensures that it is replicated and committed.
     /// </summary>
     /// <param name="cluster">The cluster node.</param>
     /// <param name="payload">The log entry payload.</param>
@@ -18,4 +21,35 @@ public static class RaftClusterExtensions
     public static ValueTask<bool> ReplicateAsync(this IRaftCluster cluster, ReadOnlyMemory<byte> payload, object? context = null,
         CancellationToken token = default)
         => cluster.ReplicateAsync<BinaryLogEntry>(new() { Content = payload, Term = cluster.Term, Context = context }, token);
+
+    /// <summary>
+    /// Appends binary log entry and ensures that it is replicated and committed.
+    /// </summary>
+    /// <typeparam name="T">The type of the binary formattable log entry.</typeparam>
+    /// <param name="cluster">The cluster node.</param>
+    /// <param name="payload">The log entry payload.</param>
+    /// <param name="context">The context to be passed to the state machine.</param>
+    /// <param name="token">The token that can be used to cancel the operation.</param>
+    /// <returns><see langword="true"/> if the appended log entry has been committed by the majority of nodes; <see langword="false"/> if retry is required.</returns>
+    /// <exception cref="InvalidOperationException">The current node is not a leader.</exception>
+    /// <exception cref="OperationCanceledException">The operation has been canceled.</exception>
+    public static ValueTask<bool> ReplicateAsync<T>(this IRaftCluster cluster, T payload, object? context = null, CancellationToken token = default)
+        where T : IBinaryFormattable<T>
+        => cluster.ReplicateAsync<BinaryLogEntry<T>>(new() { Content = payload, Term = cluster.Term, Context = context }, token);
+
+    /// <summary>
+    /// Appends JSON log entry and ensures that it is replicated and committed.
+    /// </summary>
+    /// <typeparam name="T">The type of the JSON log entry.</typeparam>
+    /// <param name="cluster">The cluster node.</param>
+    /// <param name="payload">The log entry payload.</param>
+    /// <param name="context">The context to be passed to the state machine.</param>
+    /// <param name="token">The token that can be used to cancel the operation.</param>
+    /// <returns><see langword="true"/> if the appended log entry has been committed by the majority of nodes; <see langword="false"/> if retry is required.</returns>
+    /// <exception cref="InvalidOperationException">The current node is not a leader.</exception>
+    /// <exception cref="OperationCanceledException">The operation has been canceled.</exception>
+    public static ValueTask<bool> ReplicateJsonAsync<T>(this IRaftCluster cluster, T payload, object? context = null,
+        CancellationToken token = default)
+        where T : IJsonSerializable<T>
+        => cluster.ReplicateAsync<JsonLogEntry<T>>(new() { Content = payload, Term = cluster.Term, Context = context }, token);
 }