ISubject<T> 인터페이스

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

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

통사론

'Declaration
Public Interface ISubject(Of T) _
    Inherits ISubject(Of T, T), IObserver(Of T),  _
    IObservable(Of T)

'Usage
Dim instance As ISubject(Of T)

public interface ISubject<T> : ISubject<T, T>, 
    IObserver<T>, IObservable<T>

generic<typename T>
public interface class ISubject : ISubject<T, T>, 
    IObserver<T>, IObservable<T>

type ISubject<'T> =  
    interface
        interface ISubject<'T, 'T>
        interface IObserver<'T>
        interface IObservable<'T>
    end

JScript does not support generic types and methods.

형식 매개 변수

• T
  제목 형식입니다.

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

방법

맨 위로

확장 메서드

맨 위로

발언

ISubject 인터페이스를 구현하면 사용자 지정 동작을 사용하여 사용자 고유의 주제를 만들 수 있습니다.

예제

다음 예제 코드는 콘솔 창에 메서드 실행을 에코하는 예제 제목을 구현합니다.

using System;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;

namespace Example
{
  class Program
  {
    static void Main()
    {
      //************************************************************************************//
      //*** The ConsoleEchoSubject will just echo it's method calls out to the console   ***//
      //*** window.                                                                      ***//
      //***                                                                              ***//
      //*** Create an instance of the ConsoleEchoSubject and subscribe to a sequence of  ***//
      //*** the first 5 integers returned by the Interval operator.                      ***//
      //***                                                                              ***//
      //************************************************************************************//

      ConsoleEchoSubject<long> mySubject = new ConsoleEchoSubject<long>();
      var obs = Observable.Interval(TimeSpan.FromSeconds(1), Scheduler.ThreadPool).Take(5);
      Console.WriteLine("Subscribing ConsoleEchoSubject to observable sequence.\n");
      obs.Subscribe(mySubject);


      //************************************************************************************//
      //*** Subscribe to the subject's observable interface. Each value will be written  ***//
      //*** to the console window.                                                       ***//
      //************************************************************************************//

      IDisposable subscription = mySubject.Subscribe(x => Console.WriteLine(x), () => 
      {
        Console.WriteLine("\nSequence Completed.\n");
        Console.WriteLine("Press ENTER to exit...");
      });

      Console.ReadLine();


      //***************************************//
      //*** Explicitly releasing resources. ***//
      //***************************************//

      subscription.Dispose();
      mySubject.Dispose();
    }
  }



  //************************************************************************************//
  //***                                                                              ***//
  //*** The ConsoleEchoSubject will just echo it's method calls out to the console   ***//
  //*** window.                                                                      ***//
  //***                                                                              ***//
  //************************************************************************************//

  class ConsoleEchoSubject<T> : ISubject<T>, IDisposable
  { 
    private List<IObserver<T>> observerList;     
    private bool isDisposed;
    private bool isStopped;
    object gate = new object();
    Exception exception;

    public ConsoleEchoSubject()
    {
      observerList = new List<IObserver<T>>();     
    }

    public void OnCompleted()
    {      
      //****************************************************************************************//
      //*** Make sure the OnCompleted operation is not preempted by another operation        ***//
      //*** which would break the expected behavior.  For example, don't allow an error from ***//
      //*** OnError preempt OnCompleted from anotther thread. Then OnCompleted would follow  ***//
      //*** an error.  That would be an incorrect behavior.                                  ***//
      //****************************************************************************************//

      lock (gate) 
      {
        CheckDisposed();

        if (!isStopped)
        {
          Console.WriteLine("\nConsoleEchoSubject : OnCompleted()");

          foreach (IObserver<T> observer in observerList)
          {
            observer.OnCompleted();
          }

          observerList.Clear();
          isStopped = true;
        }
      }
    }

    public void OnError(Exception error)
    {
      if (error == null)
        throw new ArgumentException("Exception error should not be null.");

      //****************************************************************************************//
      //*** Make sure the OnError operation is not preempted by another operation which      ***//
      //*** would break the expected behavior.  For example, don't allow unsubscribe or an   ***//
      //*** OnCompleted operation to preempt OnError from another thread. This would result  ***//
      //*** in an error following completion.  That would be an incorrect behavior.          ***//
      //****************************************************************************************//

      lock (gate) 
      {
        CheckDisposed();

        if (!isStopped)
        {
          Console.WriteLine("\nConsoleEchoSubject : OnError({0})", error.Message);

          exception = error;

          foreach (IObserver<T> observer in observerList)
          {
            observer.OnError(error);
          }

          observerList.Clear();
          isStopped = true;  
        }
      }
    }

    public void OnNext(T value)
    {
      //****************************************************************************************//
      //*** Make sure the OnNext operation is not preempted by another operation which       ***//
      //*** would break the expected behavior.  For example, don't allow unsubscribe, errors ***//
      //*** or an OnCompleted operation to preempt OnNext from another thread. This would    ***//
      //*** have the result of items in a sequence following completion, errors, or          ***//
      //*** unsubscribe.  That would be an incorrect behavior.                               ***//
      //****************************************************************************************//

      lock (gate)
      {
        CheckDisposed();

        if (!isStopped)
        {
          Console.WriteLine("\nConsoleEchoSubject : OnNext({0})", value.ToString());

          foreach (IObserver<T> observer in observerList)
          {
            observer.OnNext(value);
          }
        }
      }
    } 

    public IDisposable Subscribe(IObserver<T> observer)
    {
      if (observer == null)
        throw new ArgumentException("observer should not BehaviorSubject null.");

      //****************************************************************************************//
      //*** Make sure Subscribe occurs in sync with the other operations so we keep the      ***//
      //*** correct behavior depending on whether an error has occurred or the observable    ***//
      //*** sequence has completed.                                                          ***//
      //****************************************************************************************//

      lock (gate)
      {
        CheckDisposed();

        if (!isStopped)
        {
          Console.WriteLine("\nConsoleEchoSubject : Creating new subscription.");
          observerList.Add(observer);
          return new Subscription(observer, this);
        }
        else if(exception != null)
        {
          observer.OnError(exception);
          return Disposable.Empty;
        }
        else
        {
          observer.OnCompleted();
          return Disposable.Empty;
        }
      }
    }

    private void Unsubscribe(IObserver<T> observer)
    {
      //****************************************************************************************//
      //*** Make sure Unsubscribe occurs in sync with the other operations so we keep the    ***//
      //*** correct behavior.                                                                ***//
      //****************************************************************************************//

      lock (gate)
      {
        Console.WriteLine("\nConsoleEchoSubject : Unsubscribing subscription.");
        observerList.Remove(observer);
      }
    }

    public void  Dispose()
    {
      //****************************************************************************************//
      //*** Make sure Dispose occurs in sync with the other operations so we keep the        ***//
      //*** correct behavior. For example, Dispose shouldn't preempt the other operations    ***//
      //*** changing state variables after they have been checked.                           ***//
      //****************************************************************************************//

      lock (gate)
      {
        Console.WriteLine("\nConsoleEchoSubject : Disposing resources.");
        observerList.Clear();
        isStopped = true;
        isDisposed = true;
      }
    }

    private void CheckDisposed()
    {
      if (isDisposed)
        throw new ObjectDisposedException("Subject has been disposed.");
    }


    //************************************************************************************//
    //***                                                                              ***//
    //*** The Subscription class wraps each observer that creates a subscription. This ***//
    //*** is needed to expose an IDisposable interface through which a observer can    ***//
    //*** cancel the subscription.                                                     ***//
    //***                                                                              ***//
    //************************************************************************************//

    class Subscription : IDisposable
    {
      private ConsoleEchoSubject<T> subject;
      private IObserver<T> observer;

      public Subscription(IObserver<T> obs, ConsoleEchoSubject<T> sub)
      {
        subject = sub;
        observer = obs;
      }

      public void Dispose()
      {
        subject.Unsubscribe(observer);  
      }
    }
  }
}

다음 출력은 예제 코드에 의해 생성되었습니다.

Subscribing ConsoleEchoSubject to observable sequence.


ConsoleEchoSubject : Creating new subscription.

ConsoleEchoSubject : OnNext(0)
0

ConsoleEchoSubject : OnNext(1)
1

ConsoleEchoSubject : OnNext(2)
2

ConsoleEchoSubject : OnNext(3)
3

ConsoleEchoSubject : OnNext(4)
4

ConsoleEchoSubject : OnCompleted()

Sequence Completed.

Press ENTER to exit...


ConsoleEchoSubject : Unsubscribing subscription.

ConsoleEchoSubject : Disposing resources.

C:\

참고 항목

참조

system.Reactive.Subjects 네임스페이스