从进程 - 输出缓冲区问题创建iconnectablebservable

Question

下面是我尝试将system.diagnostics.process转换为iconnectablebservable。 此解决方案存在问题：我想仔细聆听标准输出和错误，并使用事件进程.Exit为OnCompleted触发。遗憾的是，我发现这个过程。在输出缓冲区为空之前提出。这意味着没有带有线程睡眠的我丑陋的解决方法，我可以通过onnext语句来重现输出输出的情况。

Q1：您是否看到此问题的任何解决方法？

q2：关于system.revive：我在解决方案中可以更好地做些什么？

问候，

markus

public static class RxProcessUtilities
{
    /// <summary>
    /// Creates a connectable observable for a process.
    /// </summary>
    /// <remarks>Must be a connectable observable in order to hinder multiple 
    /// subscriptions to call the process multiple times.</remarks>
    /// <param name="process">The process.</param>
    /// <returns></returns>
    public static IConnectableObservable<string> CreateConnectableObservableProcess
        (string filename, string arguments, IObservable<string> input = null)
    {
        var observable = Observable.Using(() =>
            {
                Process process = new Process();

                // process configuration
                process.StartInfo.FileName = filename;
                process.StartInfo.Arguments = arguments;
                process.StartInfo.CreateNoWindow = true;
                process.StartInfo.UseShellExecute = false;

                process.EnableRaisingEvents = true;
                process.StartInfo.RedirectStandardError = true;
                process.StartInfo.RedirectStandardOutput = true;

                if (null != input)
                {
                    process.StartInfo.RedirectStandardInput = true;

                    input.Subscribe(s =>
                        {
                            if (!process.HasExited)
                            {
                                process.StandardInput.Write(s);
                            }
                        });
                }

                return process;
            },
            process =>
            {
                return Observable.Create<string>(
                (IObserver<string> observer) =>
                {
                    // listen to stdout and stderr
                    var stdOut = RxProcessUtilities.CreateStandardOutputObservable(process);
                    var stdErr = RxProcessUtilities.CreateStandardErrorObservable(process);

                    var stdOutSubscription = stdOut.Subscribe(observer);
                    var stdErrSubscription = stdErr.Subscribe(observer);

                    var processExited = Observable.FromEventPattern
                    (h => process.Exited += h, h => process.Exited -= h);

                    var processError = processExited.Subscribe(args =>
                    {
                        // Here is my problem: process sends exited event *before* all 
                        // *DataReceived events have been raised

                        // My ugly workaround for process exit before stdout and stderr buffers are empty.
                        Thread.Sleep(2000);

                        // Also: AFAICS we cannot read synchronously what is left in the buffer, 
                        // since we started asynchronously. This will throw:
                        // string restOfStdOut = process.StandardOutput.ReadToEnd();
                        // string restOfStdErr = process.StandardError.ReadToEnd();

                        if (process.ExitCode != 0)
                        {
                            observer.OnError(new Exception
                                (String.Format("Process '{0}' terminated with error code {1}",
                                 process.StartInfo.FileName, process.ExitCode)));
                        }
                        else
                        {
                            observer.OnCompleted();
                        }
                    });

                    process.Start();

                    process.BeginOutputReadLine();
                    process.BeginErrorReadLine();

                    return new CompositeDisposable
                        (stdOutSubscription,
                         stdErrSubscription,
                         processError);
                });
            });

        return observable.Publish();
    }

    /// <summary>
    /// Creates an IObservable&lt;string&gt; for the standard error of a process.
    /// </summary>
    /// <param name="process">The process.</param>
    /// <returns></returns>
    public static IObservable<string> CreateStandardErrorObservable(Process process)
    {
        // var processExited = Observable.FromEventPattern
        //    (h => process.Exited += h, h => process.Exited -= h);

        var receivedStdErr =
            Observable.FromEventPattern<DataReceivedEventHandler, DataReceivedEventArgs>
                (h => process.ErrorDataReceived += h,
                 h => process.ErrorDataReceived -= h)
            //.TakeUntil(processExited) 
            // cannot be used here, since process exited event might be raised 
            // before all stderr and stdout events occurred.
            .Select(e => e.EventArgs.Data);

        return Observable.Create<string>(observer =>
        {
            var cancel = Disposable.Create(process.CancelErrorRead);

            return new CompositeDisposable(cancel, receivedStdErr.Subscribe(observer));
        });
    }

    /// <summary>
    /// Creates an IObservable&lt;string&gt; for the standard output of a process.
    /// </summary>
    /// <param name="process">The process.</param>
    /// <returns></returns>
    public static IObservable<string> CreateStandardOutputObservable(Process process)
    {
        var receivedStdOut =
            Observable.FromEventPattern<DataReceivedEventHandler, DataReceivedEventArgs>
            (h => process.OutputDataReceived += h,
             h => process.OutputDataReceived -= h)
            .Select(e => e.EventArgs.Data);

        return Observable.Create<string>(observer =>
        {
            var cancel = Disposable.Create(process.CancelOutputRead);

            return new CompositeDisposable(cancel, receivedStdOut.Subscribe(observer));
        });
    }
}

.

Answer 1

技巧是

process.WaitForExit();

.

请参阅 http://msdn.microsoft.com/en-us/library/fb4aw7b8.aspx ：“此过载[waitforexit（）]可确保完成所有处理，包括处理重定向标准输出的异步事件。您应该在调用WAITFOREXIT后使用此过载（INT32）当标准输出被重定向到异步事件处理程序时，过载。“

这是完整的解决方案：

    /// <summary>
    /// Creates a connectable observable for a process.
    /// </summary>
    /// <remarks>Must be a connectable observable in order to hinder multiple subscriptions to call the process multiple times.</remarks>
    /// <param name="process">The process.</param>
    /// <returns></returns>
    public static IConnectableObservable<string> CreateConnectableObservableProcess(string filename, string arguments, IObservable<string> input = null)
    {
        var observable = Observable.Using(() =>
            {
                Process process = new Process();

                // process configuration
                process.StartInfo.FileName = filename;
                process.StartInfo.Arguments = arguments;
                process.StartInfo.CreateNoWindow = true;
                process.StartInfo.UseShellExecute = false;

                process.EnableRaisingEvents = true;
                process.StartInfo.RedirectStandardError = true;
                process.StartInfo.RedirectStandardOutput = true;

                if (null != input)
                {
                    process.StartInfo.RedirectStandardInput = true;

                    input.Subscribe(s =>
                        {
                            if (!process.HasExited)
                            {
                                process.StandardInput.Write(s);
                            }
                        });
                }

                return process;
            },
            process =>
            {
                return Observable.Create<string>(
                    (IObserver<string> observer) =>
                    {
                        // listen to stdout and stderr
                        var stdOut = RxProcessUtilities.CreateStandardOutputObservable(process);
                        var stdErr = RxProcessUtilities.CreateStandardErrorObservable(process);

                        var stdOutSubscription = stdOut.Subscribe(observer);
                        var stdErrSubscription = stdErr.Subscribe(observer);

                        var processExited = Observable.FromEventPattern(h => process.Exited += h, h => process.Exited -= h);

                        var processError = processExited.Subscribe(args =>
                        {
                            process.WaitForExit();

                            try
                            {
                                if (process.ExitCode != 0)
                                {
                                    observer.OnError(new Exception(String.Format("Process '{0}' terminated with error code {1}",
                                        process.StartInfo.FileName, process.ExitCode)));
                                }
                                else
                                {
                                    observer.OnCompleted();
                                }
                            }
                            finally
                            {
                                process.Close();
                            }
                        });

                        process.Start();

                        process.BeginOutputReadLine();
                        process.BeginErrorReadLine();

                        return new CompositeDisposable(stdOutSubscription,
                                                       stdErrSubscription,
                                                       processError);
                    });
            });

        return observable.Publish();
    }

.