상태를 유지하지 않고 관찰 가능한 스트림에서 항목 수를 얻는 방법?

Question

수신 국 관용구를 사용하여 어떤 특정 시점에이 학교의 학생 수를 얻으려면 School 클래스에서 상태를 유지할 필요가 없습니까?

using System; 
using System.Reactive.Subjects; 

namespace SchoolManagementSystem 
{ 
    public class School 
    { 
     private ISubject<Student> _subject = null; 
     private int _maxNumberOfSeats; 
     private int _numberOfStudentsAdmitted; 

     public string Name { get; set; } 

     public School(string name, int maxNumberOfSeats) 
     { 
      Name = name; 

      _maxNumberOfSeats = maxNumberOfSeats; 

      _numberOfStudentsAdmitted = 0; 

      _subject = new ReplaySubject<Student>(); 
     } 

     public void AdmitStudent(Student student) 
     { 
      try 
      { 
       if (student == null) 
        throw new ArgumentNullException("student"); 

       if (_numberOfStudentsAdmitted == _maxNumberOfSeats) 
       { 
        _subject.OnCompleted(); 
       } 

       // Obviously can't do this because this will 
       // create a kind of dead lock in that it will 
       // wait for the _subject to complete, but I am 
       // using the same _subject to issue notifications. 

       // _numberOfStudentsAdmitted = _subject.Count().Wait(); 


       // OR to keep track of state myself 
       Interlocked.Increment(ref _numberOfStudentsAdmitted); 

       _subject?.OnNext(student); 
      } 
      catch(Exception ex) 
      { 
       _subject.OnError(ex); 
      } 
     } 

     public IObservable<Student> Students 
     { 
      get 
      { 
       return _subject; 
      } 
     } 
    } 
} 

아니면 그냥 Rx를 사용하여 설계된 구성 요소의 원칙과 연계되어 있지 않습니까?

이것은 클라이언트의 책임이어야합니다 (카운트를 얻고 onNext 처리기의 모든 부작용을 수행하기 위해)? 그리고 observables는 단순히 무언가 관심이 있다는 것을 CPU에 알리는 하드웨어 인터럽트 루틴과 마찬가지로 stateless signal-sources 또는 gate로 작동해야합니다.

이 경우 우리는 신호 완료에 대한 관찰 가능 기준을 상실합니다. 그렇다면 완성 시점을 어떻게 알 수 있습니까?

Answer 1

시퀀스에서 Count() 메서드를 사용할 수 있습니다. 그것은 관찰 된 순서를 만들어 그 각각의 값은 최신 총 학생 수를 나타냅니다. _subject.

그러면 학생 수 값 순서에 반응 할 수 있습니다. Zip() 연산은 내부 시퀀스 중 하나가 완료되면 결과 시퀀스를 완료하는 데 이점이 있으므로 연산이 유용 할 수 있으므로 TakeWhile을 사용하여 강제로 수행 할 수 있습니다.

결과는 단순히 (이미처럼) _subject?.OnNext(student)와 순서에 새 학생을 밀어하는 것 AdmitStudent 메소드 본문에서 할 남아있을 것이다이 모든

Observable.Zip(
    _subject.Select(student => != null ? student ? throw new ArgumentNullException("student")), 
    _subject.Count().TakeWhile(studentCount => studentCount < _maxNumberOfSeats), 
    (student, count) => student 
); 

같이 보입니다 만, 여분의 논리없이. 또한 최대 수정 횟수에 도달하면 _subject 그 자체가 완료되도록하기 위해 조금 수정할 수도 있지만 비즈니스 규칙에 대해서는 확실하지 않으므로 결정할 수 있도록 남겨 두겠습니다.

마지막으로 Rx 유형의 확장 기능을 사용하고 자유롭게 사용하는이 웹 사이트를 둘러 보는 것이 좋습니다.