Làm thế nào để hoàn thành một Rx Observable tùy thuộc vào một điều kiện trong một sự kiện

Question

Tôi có một sự kiện mà tôi không kiểm soát được trong đó cung cấp cho tôi dữ liệu. EventArgs trông giống như sau:

class MyEventArg { 
    bool IsLastItem {get;} 
    Data DataItem {get;} 
} 

Tôi sử dụng Rx để chuyển sự kiện này thành IObservable. Nhưng tôi muốn hoàn thành việc quan sát nếu IsLastItem là sự thật.

Bất kỳ ý tưởng thanh lịch nào? Có một cách để phân loại dữ liệu thông qua một chủ đề mà tôi có nhiều quyền kiểm soát hơn để đặt sự kiện OnComplete nếu điều kiện xảy ra ...

Answer 1

Nếu bạn muốn phần tử cuối cùng được đưa vào, bạn có thể hợp nhất luồng chỉ với phần tử cuối cùng cùng với luồng thông thường kết hợp với TakeWhile. Đây là một ứng dụng giao diện điều khiển đơn giản để chứng minh điều đó:

var subject = new List<string> 
{        
"test", 
"last" 
}.ToObservable(); 

var my = subject 
      .Where(x => x == "last").Take(1) 
      .Merge(subject.TakeWhile(x => x != "last")); 

my.Subscribe(
    o => Console.WriteLine("On Next: " + o), 
    () => Console.WriteLine("Completed")); 

Console.ReadLine(); 

in này:

On Next: test 
On Next: last 
Completed 

CẬP NHẬT Có một lỗi mà supressed thông điệp OnCompleted nếu Quan sát cơ bản không thực sự hoàn chỉnh . Tôi sửa lại đoạn code để đảm bảo OnCompleted được gọi

Và nếu bạn muốn tránh đăng ký vào các chuỗi cơ bản nhiều lần cho quan sát lạnh bạn có thể cấu trúc lại các mã như thế này:

var my = subject.Publish(p => p 
      .Where(x => x == "last").Take(1) 
      .Merge(p.TakeWhile(x => x != "last"))); 

Answer 2

Bạn đang tìm kiếm một cái gì đó như thế này?

IObservable<MyEventArg> result = 
    myEventArgObservable.TakeWhile(arg => !arg.IsLastItem); 

Answer 3

public static IObservable<TSource> TakeWhileInclusive<TSource>(
     this IObservable<TSource> source, Func<TSource, bool> predicate) 
{ 
    return Observable 
     .Create<TSource>(o => source.Subscribe(x => 
                { 
                 o.OnNext(x); 
                 if (!predicate(x)) 
                  o.OnCompleted(); 
                }, 
               o.OnError, 
               o.OnCompleted 
           )); 
}