Reactive Extensions:使用适用于文件的Rx创建管道

Question

我有一个包含3个步骤的流程管道:

1. 视频到图像:我有一个视频转换为静止图像(帧)
2. zip文件的帧:当视频中的所有帧都已处理完毕后,我应该用它们创建一个Zip文件.
3. zip文件=>上传到FTP

它涉及2个一次性用品:视频捕获和zip文件.

我怎么能用Rx处理它？有任何想法吗？很抱歉没有发布任何代码,我不知道如何开始.

提前致谢!

Answer 1

您是否必须从每个步骤传递原始对象.视频捕获或zip文件被"处理"应该没问题,因为我确信动作会产生一些副作用(a MemoryStream,写入磁盘的文件等).您可以将指针(Uri的？)传递给每个操作的结果到管道的下一部分？

void Main()
{
    var video = new Uri("https://www.youtube.com/watch?v=Tp5mRlHwZ7M");
    var query = from frames in TranscodeVideoToImages(video)
        from zipFile in ZipFiles(frames)
        from uploadLocation in UploadFile(zipFile)
        select uploadLocation;
    query.Subscribe(...)
}
private IObservable<Uri[]> TranscodeVideoToImages(Uri imageSource)
{
    //Do some long running (async) work here.
    // Save work to disk
    // Return location of saved work
}
private IObservable<Uri> ZipFiles(Uri[] files)
{
    //Run the zip process on all of the files
    //  Return the location of the zip file
}
private IObservable<Uri> UploadFile(Uri source)
{
    //Upload the File. 
    //Probably as simple as as task based operation with .ToObservable()
}