RX 제목, 조건이 충족 될 때까지 그룹화/연결

Question

항목 스트림이 있고 조건이 충족 될 때까지 속성별로 그룹화하는 구독을 만들고 싶습니다.

예를 들어 Id이 나타날 때까지 EventItem을 이름으로 그룹화한다고 가정 해 봅시다. 요점은 더 이상 관련 이벤트가 없다는 것을 알리는 특정 Id을 볼 때까지 이벤트를 상관시키는 것입니다. 그런 다음 상호 연관된 이벤트의 각 그룹에 대해 몇 가지 작업을 수행하려고합니다.

public class EventItem 
{ 
    public int Id { get; set } 
    public string Name { get; set; } 
} 

// Using a Subject since it seems the simplest way 
Subject<EventItem> eventStream; 
... 

// A seperate thread pushes EventItem objects into the Subject 
eventStream.OnNext(eventItem); 

// Struggling here... 
IDisposable subscription = eventStream.???? 

나는 TakeWhile 등 TakeUntil, GroupByUntil, GroupBy으로 여러 조합을 시도했습니다,하지만 난

Answer 1

편집 (I은 수신과 함께 오히려 경험이 있어요)이 작업을 수행하는 방법을 알아낼 수 없습니다 @ 바탕으로

슐 로모의 대답하고 싶은에 대한 영업 이익의 의견을 완료 할 수 있습니다

var subscription = producer 
     .TakeWhile(ei => ei.Id != 42) 
     .GroupBy(x => x.Name) 
     .Select(o => o) 
     .SelectMany(o => o.ToList()) //If you want y to be IList<EventItem>, use this line. If you prefer y to be IObservable<EventItem>, comment out. 
     .Repeat() 
     .Subscribe(); 

,

원본 응답

이 방법이 유용합니까?

 Subject<Notification> producer = new Subject<Notification>(); 

     //This way there's only one producer feeding the group and the duration 
     var connectedProducer = 
      producer 
       .Publish() 
       .RefCount(); 

     connectedProducer 
      .GroupByUntil(
       item => item.Name, 
       item => connectedProducer.Where(x=> x.Id == 3)) 
      .SelectMany(result => 
      { 
       return result.Aggregate<Notification, List<Notification>>(new List<Notification>(), (dict, item) => 
       { 
        dict.Add(item); 
        return dict; 
       }); 
      }) 
      //not sure if you need this but just a way to filter out the signal 
      .Where(item => item.First().Id != 3) 
      .Subscribe(results => 
      { 
       //This will only run after a "3" is passed in and then you get a list of everything passed in with the names 
       //not sure if you wanted intermediate results or for it all to just come through once the signal indicates processing 
      }); 

저는 이전 답변에서 벗어나 약간 수정했습니다. Rx grouped throttling

Answer 2

이미 알고있는 답변이 있지만 복잡한 것 같습니다. Enigmativity의 코멘트 @ 떨어져 사용 :

var subscription = eventStream 
    .GroupBy(x => x.Name) 
    .Select(o => o.TakeWhile(ei => ei.Id != 42)) 
    .SelectMany(o => o.ToList()) //If you want y to be IList<EventItem>, use this line. If you prefer y to be IObservable<EventItem>, comment out. 
    .Subscribe(y => {});