ReplaySubject<T> 클래스

관찰 가능한 시퀀스인 개체와 관찰자를 나타냅니다.

상속 계층 구조

System.Object
  System.Reactive.Subjects.ReplaySubject<T>

네임스페이스:System.Reactive.Subjects
어셈블리: System.Reactive(System.Reactive.dll)

통사론

'Declaration
Public NotInheritable Class ReplaySubject(Of T) _
    Implements ISubject(Of T), ISubject(Of T, T),  _
    IObserver(Of T), IObservable(Of T), IDisposable

'Usage
Dim instance As ReplaySubject(Of T)

public sealed class ReplaySubject<T> : ISubject<T>, 
    ISubject<T, T>, IObserver<T>, IObservable<T>, IDisposable

generic<typename T>
public ref class ReplaySubject sealed : ISubject<T>, 
    ISubject<T, T>, IObserver<T>, IObservable<T>, IDisposable

[<SealedAttribute>]
type ReplaySubject<'T> =  
    class
        interface ISubject<'T>
        interface ISubject<'T, 'T>
        interface IObserver<'T>
        interface IObservable<'T>
        interface IDisposable
    end

JScript does not support generic types and methods.

형식 매개 변수

• T
  형식입니다.

ReplaySubject<T> 형식은 다음 멤버를 노출합니다.

생성자

맨 위로

방법

맨 위로

확장 메서드

맨 위로

발언

주체는 구독자 및 게시자 역할을 한다는 측면에서 프록시와 유사하게 작동합니다. 해당 IObserver 인터페이스를 사용하여 관찰 가능한 여러 데이터 시퀀스를 구독할 수 있습니다. 그런 다음 데이터는 주체의 IObservable 인터페이스를 통해 게시됩니다.

ReplaySubject는 수신하는 항목을 버퍼링합니다. 따라서 나중에 만든 구독은 이미 게시된 경우에도 버퍼링된 시퀀스의 항목에 액세스할 수 있습니다. ReplaySubject가 버퍼링할 항목 수는 생성자에 전달된 매개 변수에 따라 달라집니다.

일반 주체는 스케줄러를 사용하여 구독된 관찰자에게 나가는 호출을 동기화합니다.

예제

이 예제에서는 관찰 가능한 문자열 시퀀스의 형태로 모의 뉴스 피드인 NewsHeadlineFeed 클래스를 만들었습니다. Generate 연산자를 사용하여 3초 이내에 임의 뉴스 헤드라인을 지속적으로 생성합니다.

ReplaySubject는 NewsHeadlineFeed 클래스의 두 뉴스 피드를 구독하기 위해 만들어집니다. 제목이 피드를 구독하기 전에 타임스탬프 연산자를 사용하여 각 헤드라인을 타임스탬프합니다. 따라서 ReplaySubject가 실제로 구독하는 시퀀스는 IObservable<타임스탬프<문자열>>형식입니다. 헤드라인 시퀀스 타임스탬프를 사용하면 구독자는 주체의 관찰 가능한 인터페이스를 구독하여 타임스탬프에 따라 데이터 스트림 또는 스트림의 하위 집합을 관찰할 수 있습니다.

ReplaySubject는 수신하는 항목을 버퍼링합니다. 따라서 나중에 만든 구독은 이미 버퍼링되고 게시된 시퀀스의 항목에 액세스할 수 있습니다. 구독은 지역 뉴스 구독이 생성되기 10초 전에 발생한 지역 뉴스 헤드라인만 수신하는 ReplaySubject에 만들어집니다. 그래서 우리는 기본적으로 ReplaySubject "재생"10 초 전에 무슨 일이 있었는지있다.

지역 뉴스 헤드라인에는 newsLocation 부분 문자열("해당 지역")만 포함되어 있습니다.

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Concurrency;
using System.Threading;

namespace Example
{
  class Program
  {
    static void Main()
    {
      //*****************************************************************************************************//
      //*** A subject acts similar to a proxy in that it acts as both a subscriber and a publisher        ***//
      //*** It's IObserver interface can be used to subscribe to multiple streams or sequences of data.   ***//
      //*** The data is then published through it's IObservable interface.                                ***//
      //***                                                                                               ***//
      //*** In this example a simple ReplaySubject is used to subscribe to multiple news feeds            ***//
      //*** that provide random news headlines. Before the subject is subscribed to the feeds, we use     ***//
      //*** Timestamp operator to timestamp each headline. Subscribers can then subscribe to the subject  ***//
      //*** observable interface to observe the data stream(s) or a subset of the stream(s) based on      ***//
      //*** time.                                                                                         ***//
      //***                                                                                               ***//
      //*** A ReplaySubject buffers items it receives. So a subscription created at a later time can      ***//
      //*** access items from the sequence which have already been published.                             ***//
      //***                                                                                               ***//
      //*** A subscriptions is created to the ReplaySubject that receives only local news headlines which ***//
      //*** occurred 10 seconds before the local news subscription was created. So we basically have the  ***//
      //*** ReplaySubject "replay" what happened 10 seconds earlier.                                      ***//
      //***                                                                                               ***//
      //*** A local news headline just contains the newsLocation substring ("in your area.").             ***//
      //***                                                                                               ***//
      //*****************************************************************************************************//

      ReplaySubject<Timestamped<string>> myReplaySubject = new ReplaySubject<Timestamped<String>>();


      //*****************************************************************//
      //*** Create news feed #1 and subscribe the ReplaySubject to it ***//
      //*****************************************************************//

      NewsHeadlineFeed NewsFeed1 = new NewsHeadlineFeed("Headline News Feed #1");
      NewsFeed1.HeadlineFeed.Timestamp().Subscribe(myReplaySubject);


      //*****************************************************************//
      //*** Create news feed #2 and subscribe the ReplaySubject to it ***//
      //*****************************************************************//

      NewsHeadlineFeed NewsFeed2 = new NewsHeadlineFeed("Headline News Feed #2");
      NewsFeed2.HeadlineFeed.Timestamp().Subscribe(myReplaySubject);


      //*****************************************************************************************************//
      //*** Create a subscription to the subject's observable sequence. This subscription will filter for ***//
      //*** only local headlines that occurred 10 seconds before the subscription was created.            ***//
      //***                                                                                               ***//
      //*** Since we are using a ReplaySubject with timestamped headlines, we can subscribe to the        ***//
      //*** headlines already past. The ReplaySubject will "replay" them for the localNewSubscription     ***//
      //*** from its buffered sequence of headlines.                                                      ***//
      //*****************************************************************************************************//

      Console.WriteLine("Waiting for 10 seconds before subscribing to local news headline feed.\n");
      Thread.Sleep(10000);

      Console.WriteLine("\n*** Creating local news headline subscription at {0} ***\n", DateTime.Now.ToString());
      Console.WriteLine("This subscription asks the ReplaySubject for the buffered headlines that\n" +
                        "occurred within the last 10 seconds.\n\nPress ENTER to exit.", DateTime.Now.ToString());

      DateTime lastestHeadlineTime = DateTime.Now;
      DateTime earliestHeadlineTime = lastestHeadlineTime - TimeSpan.FromSeconds(10);     
      
      IDisposable localNewsSubscription = myReplaySubject.Where(x => x.Value.Contains("in your area.") &&
                                                               (x.Timestamp >= earliestHeadlineTime) &&
                                                               (x.Timestamp < lastestHeadlineTime)).Subscribe(x =>
      {
        Console.WriteLine("\n************************************\n" +
                          "***[ Local news headline report ]***\n" +
                          "************************************\n" + 
                          "Time         : {0}\n{1}\n\n", x.Timestamp.ToString(), x.Value);
      });

      Console.ReadLine();


      //*******************************//
      //*** Cancel the subscription ***//
      //*******************************//

      localNewsSubscription.Dispose();
      

      //*************************************************************************//
      //*** Unsubscribe all the ReplaySubject's observers and free resources. ***//
      //*************************************************************************//

      myReplaySubject.Dispose();    
    }
  }



  //*********************************************************************************//
  //***                                                                           ***//
  //*** The NewsHeadlineFeed class is just a mock news feed in the form of an     ***//
  //*** observable sequence in Reactive Extensions.                               ***//
  //***                                                                           ***//
  //*********************************************************************************//
  class NewsHeadlineFeed
  {
    private string feedName;                     // Feedname used to label the stream
    private IObservable<string> headlineFeed;    // The actual data stream
    private readonly Random rand = new Random(); // Used to stream random headlines.


    //*** A list of predefined news events to combine with a simple location string ***//
    static readonly string[] newsEvents = { "A tornado occurred ",
                                            "Weather watch for snow storm issued ",
                                            "A robbery occurred ",
                                            "We have a lottery winner ",
                                            "An earthquake occurred ",
                                            "Severe automobile accident "};

    //*** A list of predefined location strings to combine with a news event. ***//
    static readonly string[] newsLocations = { "in your area.",
                                               "in Dallas, Texas.",
                                               "somewhere in Iraq.",
                                               "Lincolnton, North Carolina",
                                               "Redmond, Washington"};

    public IObservable<string> HeadlineFeed
    {
      get { return headlineFeed; }
    }

    public NewsHeadlineFeed(string name)
    {
      feedName = name;

      //*****************************************************************************************//
      //*** Using the Generate operator to generate a continous stream of headline that occur ***//
      //*** randomly within 5 seconds.                                                        ***//
      //*****************************************************************************************//
      headlineFeed = Observable.Generate(RandNewsEvent(),
                                         evt => true,
                                         evt => RandNewsEvent(),
                                         evt => { Thread.Sleep(rand.Next(3000)); return evt; },
                                         Scheduler.ThreadPool);
    }


    //****************************************************************//
    //*** Some very simple formatting of the headline event string ***//
    //****************************************************************//
    private string RandNewsEvent()
    {
      return "Feedname     : " + feedName + "\nHeadline     : " + newsEvents[rand.Next(newsEvents.Length)] +
             newsLocations[rand.Next(newsLocations.Length)];
    }
  }
}

다음 출력은 예제 코드와 함께 생성되었습니다. 새 피드는 임의이므로 지역 뉴스 헤드라인을 보려면 두 번 이상 실행해야 할 수 있습니다.

Waiting for 10 seconds before subscribing to local news headline feed.

** 2011년 5월 9일 오전 4:07:48에 지역 뉴스 헤드라인 구독 만들기 **

This subscription asks the ReplaySubject for the buffered headlines that
occurred within the last 10 seconds.

Press ENTER to exit.

********************************** [ 지역 뉴스 헤드 라인 보고서]********************************** 시간 : 5/9/2011 4:07:42 오전 -04:00 피드 이름 : 헤드 라인 뉴스 피드 #2 헤드 라인 : 우리는 당신의 지역에 복권 당첨자를 가지고있다.

********************************** [ 지역 뉴스 헤드 라인 보고서]********************************** 시간 : 5/9/2011 4:07:47 오전 -04:00 피드 이름 : 헤드 라인 뉴스 피드 #1 헤드 라인 : 귀하의 지역에서 발행 된 눈 폭풍에 대한 날씨 시계.

스레드 안전성

이 형식의 모든 공용 정적(Visual Basic에서 공유) 멤버는 스레드로부터 안전합니다. 모든 인스턴스 멤버는 스레드로부터 안전하게 보호되지 않습니다.

참고 항목

참조

system.Reactive.Subjects 네임스페이스