How to loop Observable Rx until condition met?

Question

I found this excellent answer that is very similar to the problem I'm trying to solve:

Process sometimes hangs while waiting for Exit

My first time working with this System.Reactive package though so I'm struggling with the usage and syntax. I'm trying to modify this block to suit my needs:

     var processExited =
            // Observable will tick when the process has gracefully exited.
            Observable.FromEventPattern<EventArgs>(process, nameof(Process.Exited))
                // First two lines to tick true when the process has gracefully exited and false when it has timed out.
                .Select(_ => true)
                .Timeout(TimeSpan.FromMilliseconds(processTimeOutMilliseconds), Observable.Return(false))
                // Force termination when the process timed out
                .Do(exitedSuccessfully => { if (!exitedSuccessfully) { try { process.Kill(); } catch {} } } );

I would like this to when it times out it checks the stdOut buffer for a particular string, if it finds it then exit, otherwise continue until the next timeout. But also, I only want to do so many timeouts before I 'give up' and kill the process so in the loop, if it doesn't exit, i would increment a counter and check that. Thank you for the help

Adding clarification, I want something like this:

int timeoutCount = 0;
const int maxTimeoutCount =5;

then in the observable, something like .Do(exitedSuccessfully => {

          if (!exitedSuccessfully) {
                        try {
                            if (timeoutCount >= maxTimeOutCount || output.ToString().Contains("string_i-m_looking_for_in_output_to_trigger_exit")) {
                                process.Kill();
                            }
                            timeOutCount++;
                        }
                        catch { }
                    }

@Asti No i mean the Timeout as shown in the code snippet I posted above. And no, I don't intend to restart the process — snappymcsnap, May 20 '20 at 17:18
" I only want to do so many timeouts before I 'give up' and kill the process so in the loop" - is what I wanted to clarify. — Asti, May 20 '20 at 17:20
In the code, you kill the process on the first timeout. Let's say you kill the process on the third timeout. What's supposed on the first timeout? — Asti, May 20 '20 at 17:22
The `Timeout` operator will trigger only once. It either passed an `OnError` or shunts the other observable in. — Asti, May 20 '20 at 18:38
@Asti So do you know of a way to make it do what I'm trying to do or is it impossible? — snappymcsnap, May 20 '20 at 19:51

Asti · Answer 1 · 2020-06-15T05:50:01.977

1

A fundamental design concept of reactive is to avoid external state - and express behavior declaratively. In Rx, you might need to model traditionally mutating state differently.

An idiomatic way to model a process as an observable is to link the lifetime of the process to subscription to the observable. Subscribing starts the process, unsubscribing stops the process, and vice versa, the process exiting normally completes the observable.

I've made an implementation of this model on Github. It's a single file library for C# and F# which abstracts the std in/out of a process as an observable.

Using that model:

 StdioObservable
    .Create("process.exe")
    .TakeUntil(line => line.Contains("Done")) // unsub when you match
    .Timeout(TimeSpan.FromSeconds(30)) // no output for 30 seconds
    .Catch((Exception exn) => Observable.Empty<string>()) //handle timeout or other
    .Subscribe();

If you want to kill the process within a duration, in spite of it producing some output, use .Take(TimeSpan.FromSeconds(30)) instead.

edited Jun 15 '20 at 05:50

answered May 20 '20 at 20:14

Asti

12,447
29
38

doesn't compile: "Cannot convert lambda expression to intended delegate type because some of the return types in the block are not implicitly convertible to the delegate return type" – snappymcsnap May 21 '20 at 17:23
Compiles fine for me. Which line, col is it? – Asti May 21 '20 at 17:34
1

I'd be inclined to change the phrase "avoid state" to "avoid **external** state". You can certainly encapsulate state in a `Create` or `Defer` observable. – Enigmativity Jun 15 '20 at 00:54
1

@Enigmativity Edited. – Asti Jun 15 '20 at 05:50

How to loop Observable Rx until condition met?

1 Answers1