Pipeline abstraction

[ruccho]

New abstraction layer of encoding pipeline to implement pipelines such as #39 and #40, which are slightly different from current one.

Concept

With the pipeline abstraction, a pipeline is represented as a sequence of pipeline elements which implement IAsyncPipelineInput<T> or IPipelineInput<T>. This enhances the reusability of the code implementing the pipeline.

IAsyncPipelineInput<T>

Represents single element of encoding pipeline, which is mainly for data flow where back pressure occurs.

IPipelineInput<T> will receive values, manipulate them and pass them another IPipelineInput<T>.

internal interface IAsyncPipelineInput<in T> : IDisposable
{
    ValueTask PushAsync(T value);
    ValueTask CompleteAsync(Exception exception = null);
}

IPipelineInput<T>

A synchronous version of IAsyncPipelineInput<T>, which is mainly for data flow without back pressure.

internal interface IPipelineInput<in T> : IDisposable
{
    void Push(T value);
    ValueTask CompleteAsync(Exception exception = null);
}

IPipelineTransform<TIn, TOut>

Some pipeline elements may be able to be used in both asynchronous and synchronous flow. IPipelineTransform only requires simple value transform implementation and they can be converted to either IAsyncPipelineInput<T> or IPipelineInput<T> using PipelineExtensions.AsInput() and PipelineExtensions.AsAsyncInput().

internal interface IPipelineTransform<in TIn, TOut> : IDisposable
{
    bool Transform(TIn input, out TOut output);
}

Redesigning RealtimeInstantReplaySession

RealtimeInstantReplaySession is redesigned with the pipeline abstraction. The pipeline is wrote like:

_videoPipeline = new FrameProviderSubscription(frameProvider, disposeFrameProvider,
    new VideoTemporalAdjuster<IFrameProvider.Frame>(_temporalController, fixedFrameInterval).AsInput(
        new FramePreprocessorInput(preprocessor, true).AsInput(
            new AsyncGPUReadbackTransform().AsInput(
                new DroppingChannelInput<LazyVideoFrameData>(
                    options.VideoInputQueueSize,
                    onLazyVideoFrameDataDropped,
                    new VideoEncoderInput(videoEncoder,
                        new BoundedEncodedDataBufferVideoInput(buffer).AsAsync()))))));

_audioPipeline = new AudioSampleProviderSubscription(audioSampleProvider, disposeAudioSampleProvider,
    new AudioTemporalAdjuster(
        _temporalController,
        options.AudioOptions.SampleRate,
        options.AudioOptions.Channels).AsInput(
        new DroppingChannelInput<PcmAudioFrame>(options.AudioInputQueueSize, onPcmAudioFrameDropped,
            new AudioEncoderInput(audioEncoder, options.AudioOptions.SampleRate,
                new BoundedEncodedDataBufferAudioInput(buffer).AsAsync()))));

It has already been confirmed that General Unbounded Recording (#40) can be implemented with a similarly small amount of code with the pipeline abstraction. (PR).

Future works

Currently all pipeline abstraction APIs are private but I want to publish them, enabling users to create their own pipelines easily.

[hkmt-mmy]

ネストが深いのが気になりました。内側のtry-finallyは関数に切り出してもいいかも？

[hkmt-mmy]

かなり長いので (もし良い感じの分割単位があれば) 分割したほうが見通しが良さそうだと思いました

[hkmt-mmy]

1行の場合に{}付けてないのにここは付いてます

[hkmt-mmy]

こことかUnregisterのところも{}ついてるけど、基本的に付けないルールでやってそうなので気になりました

[hkmt-mmy]

借りた当人じゃないのにReturnするのが少し気になりました。コンストラクタ引数でrendとしてるしいいのかな……？

[ruccho]

この配列はタスク間を非同期的にたらい回されたあとに解放されるので、残念ながらこれについては解決策がないです

[hkmt-mmy]

ネストがめっちゃ深い……。処理自体は少ないけど、private関数なりローカル関数なりに切り出したほうが良さそう

[hkmt-mmy]

readonlyがついててコンストラクタ初期化されるフィールドがまとまってないのが気になりました。好みの問題かもですが