MoShiro/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs

using System.Net;
using MoShiro.Protocol;

namespace MoShiro.Measurement;

/// <summary>积木的公共部分：环境、参数、把行切成列块交出去。</summary>
public abstract class BlockBase : IBlock
{
    protected BlockBase(BlockEnv env, BlockSpec spec)
    {
        Env = env;
        Spec = spec;
    }

    public virtual long Produced => 0;

    protected BlockEnv Env { get; }

    protected BlockSpec Spec { get; }

    public abstract Task RunAsync(CancellationToken ct);

    public virtual int Flush(FlushMode mode, long peerProduced = 0) => 0;

    public virtual RowStore? Records(string streamName) => null;

    protected IPEndPoint PeerEndpoint(int port) => new(Env.PeerAddress!, port);

    /// <summary>交出一个流的行；起点和时长取自时间列 <paramref name="timeColumn"/> 的首尾（本侧的单调时钟，微秒）。</summary>
    protected void Ship(RawStream stream, string timeColumn, IReadOnlyList<RawStream>? extra = null, IReadOnlyDictionary<string, System.Text.Json.JsonElement>? parameters = null)
    {
        RawColumn? time = stream.Find(timeColumn);
        long first = stream.Count > 0 && time is not null ? time.Values[0] : Env.Clock.NowUs();
        long last = stream.Count > 0 && time is not null ? time.Values[stream.Count - 1] : first;
        List<RawStream> streams = [stream];
        if (extra is not null)
        {
            streams.AddRange(extra);
        }

        Env.Fragments!.Ship(Spec.KindId, Spec.Scope, streams, parameters ?? Spec.Params.Values, Env.Clock.UtcMsAt(first), (int)((last - first) / 1000), Spec.Port);
    }

    /// <summary>什么也不做、一直等到取消（监听方里不需要自己运行逻辑的积木）。</summary>
    protected async Task WaitUntilCancelledAsync(CancellationToken ct)
    {
        try
        {
            await Task.Delay(Timeout.InfiniteTimeSpan, Env.Time, ct);
        }
        catch (OperationCanceledException)
        {
            // 被停止
        }
    }
}

/// <summary>
/// 拨号方作为行的主导方时记录的行：行按键（序号）追加，回复到达时按键回填列，到点把已经稳定的行从游标处切下来交出去。
/// 环形缓冲被覆盖的行（交出太慢）直接跳过。
/// </summary>
internal sealed class DriverLog
{
    private readonly object _gate = new();
    private long _cursor;

    public DriverLog(StreamSpec spec, Side driver, int capacity)
    {
        Store = new RowStore(spec, driver, capacity);
        Spec = spec;
    }

    public RowStore Store { get; }

    public StreamSpec Spec { get; }

    public long Count => Store.NextKey;

    /// <summary>取出游标之后已经稳定的行：<paramref name="settled"/> 为 null 或 <paramref name="all"/> 时取全部。</summary>
    public RawStream? Take(Func<long, bool>? settled, bool all, int maxRows = int.MaxValue)
    {
        long from;
        long to;
        lock (_gate)
        {
            from = _cursor;
            to = from;
            long end = Store.NextKey;
            while (to < end && to - from < maxRows && (all || settled is null || !Store.Contains(to) || settled(to)))
            {
                to++;
            }

            if (to == from)
            {
                return null;
            }

            _cursor = to;
        }

        RawStream stream = Store.Cut(from, to, includeKey: Spec.Key is not null);
        return stream.Count == 0 ? null : stream;
    }
}