Log Interface

log.cs

using System;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Threading.Tasks.Sources;

namespace RockR.Storage
{
    public class WriteAheadLog
    {
        private Channel<QueueItem> _workQueue = Channel.CreateUnbounded<QueueItem>(new System.Threading.Channels.UnboundedChannelOptions()
        {
            SingleReader = true,
            SingleWriter = false,
            AllowSynchronousContinuations = false,
        });

        private ChannelWriter<QueueItem> Writer => _workQueue.Writer;

        public ValueTask LogDataWaitForQueuedAsync(ReadOnlyMemory<byte> data)
        {
            var queueItem = new QueueItem() { Memory = data };
            return Writer.WriteAsync(queueItem);
        }

        public ValueTask LogDataWaitForCommitAsync(ReadOnlyMemory<byte> data)
        {
            var taskReset = new ValueTaskSource();
            var queueItem = new QueueItem() { Memory = data, Source = taskReset };
            if(Writer.TryWrite(queueItem))
            {
                return new ValueTask(taskReset, 0);
            }

            return new ValueTask(LoadDataWaitForQueuedAsync(queueItem));
            
            async Task LoadDataWaitForQueuedAsync(QueueItem queueItem)
            {
                await Writer.WriteAsync(queueItem);
                await (new ValueTask(queueItem.Source, 0));
            }
        }

        private struct QueueItem
        {
            public ReadOnlyMemory<byte> Memory;
            public ValueTaskSource Source;
        }

        private class ValueTaskSource : IValueTaskSource
        {
            private ManualResetValueTaskSourceCore<bool> _source = new ManualResetValueTaskSourceCore<bool>();

            public void GetResult(short token)
            {
            }

            public ValueTaskSourceStatus GetStatus(short token)
            {
                throw new NotImplementedException();
            }

            public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
            {
                throw new NotImplementedException();
            }
        }
    }
}