ISubject<TSource, TResult> 인터페이스

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

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

통사론

'Declaration
Public Interface ISubject(Of In TSource, Out TResult) _
    Inherits IObserver(Of TSource), IObservable(Of TResult)

'Usage
Dim instance As ISubject(Of In TSource, Out TResult)

public interface ISubject<in TSource, out TResult> : IObserver<TSource>, 
    IObservable<TResult>

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

type ISubject<'TSource, 'TResult> =  
    interface
        interface IObserver<'TSource>
        interface IObservable<'TResult>
    end

JScript does not support generic types and methods.

형식 매개 변수

• inTSource
  원본의 형식입니다.

  이 형식 매개 변수는 반공변성입니다. 즉, 지정한 형식 또는 파생되지 않은 형식을 사용할 수 있습니다. 공변성 및 반공변에 대한 자세한 내용은 참조하세요.

• outTResult
  결과의 형식입니다.

  이 형식 매개 변수는 공변성입니다. 즉, 지정한 형식 또는 더 파생된 형식을 사용할 수 있습니다. 공변성 및 반공변에 대한 자세한 내용은 참조하세요.

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

방법

맨 위로

확장 메서드

맨 위로

발언

이 주체 인터페이스는 다른 형식의 관찰 가능한 시퀀스를 게시하는 동안 한 형식의 관찰 가능한 시퀀스를 관찰할 수 있는 유연성을 주체에 제공합니다.

예제

이 예제에서는 다른 형식의 관찰 가능한 시퀀스를 게시하는 동안 한 형식의 관찰 가능한 시퀀스를 관찰하는 ISubject<TSource, TResult> 구현하는 방법을 보여 줍니다. 이 예제의 AsciiConverterSubject는 char 형식의 시퀀스를 관찰하고 관찰 가능한 int 시퀀스를 게시하여 구현을 보여 줍니다. 게시된 관찰 가능한 int 시퀀스는 관찰되는 각 문자 값에 대한 ASCII 코드입니다.

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

namespace Example
{
  class Program
  {
    static void Main()
    {
      //****************************************************************************************//
      //*** Create an observable sequence of char from console input until enter is pressed. ***//
      //****************************************************************************************//
      IObservable<char> keySequence = Observable.Create<char>(observer =>
      {
        bool bContinue = true;

        while (bContinue)
        {
          ConsoleKeyInfo keyInfo = Console.ReadKey(true);

          if (keyInfo.Key != ConsoleKey.Enter)
          {
            observer.OnNext(keyInfo.KeyChar);
          }
          else
          {
            observer.OnCompleted();
            bContinue = false;
          }
        }

        return (() => { });
      });


      //****************************************************************************************//
      //*** Create an AsciiConverterSubject which takes a source type of char and returns an ***//
      //*** observable sequence of int which is the ASCII code for source items of char.     ***//
      //****************************************************************************************//

      AsciiConverterSubject myConverterSubject = new AsciiConverterSubject();


      //****************************************************************************************//
      //*** Subscribe to the keySequence on the .NET threadpool so the main thread can       ***//
      //*** create subscriptions to the AsciiConverterSubject                                ***//
      //****************************************************************************************//

      IDisposable subscription = keySequence.SubscribeOn(Scheduler.ThreadPool).Subscribe(myConverterSubject);


      Console.WriteLine("\nEnter a sequence of keys to have the AsciiConverterSubject\nconvert the keys to their ASCII code values.\n"); 
      Console.WriteLine("Press ENTER to terminate the observable sequence...\n");


      //****************************************************************************************//
      //*** Subscribe to the AsciiConverterSubject and write the ASCII code values to the    ***//
      //*** console window.                                                                  ***//
      //***                                                                                  ***//
      //*** The main thread will wait on the completion of the keySequence. It completes     ***//
      //*** when ENTER is pressed.                                                           ***//
      //****************************************************************************************//

      EventWaitHandle waitHandle = new EventWaitHandle(false, EventResetMode.ManualReset);
      myConverterSubject.Subscribe(c => Console.WriteLine("Ascii Char code {0} entered.",c), () => waitHandle.Set());
      waitHandle.WaitOne();


      //***************************************//
      //*** Explicitly releasing resources. ***//
      //***************************************//
      
      subscription.Dispose();
      myConverterSubject.Dispose();
    }
  }



  //***********************************************************************************************//
  //***                                                                                         ***//
  //*** The AsciiConverterSubject demonstrates an implementation of ISubject<TSource, TResult>. ***//
  //*** It is used to subscribe to an observable sequence of char. It publishes an observable   ***//
  //*** sequence of int which should be the ASCII code for each char value it observes.         ***//
  //***                                                                                         ***//
  //***********************************************************************************************//

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

    public AsciiConverterSubject()
    {
      observerList = new List<IObserver<int>>();
    }

    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)
        {
          foreach (IObserver<int> 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)
        {
          exception = error;

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

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

    public void OnNext(char 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)
        {
          foreach (IObserver<int> observer in observerList)
          {
            observer.OnNext(Convert.ToInt32(value));
          }
        }
      }
    }

    public IDisposable Subscribe(IObserver<int> 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)
        {
          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<int> observer)
    {
      //****************************************************************************************//
      //*** Make sure Unsubscribe occurs in sync with the other operations so we keep the    ***//
      //*** correct behavior.                                                                ***//
      //****************************************************************************************//

      lock (gate)
      {
        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)
      {
        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 AsciiConverterSubject subject;
      private IObserver<int> observer;

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

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

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

Enter a sequence of keys to have the AsciiConverterSubject
convert the keys to their ASCII code values.

Press ENTER to terminate the observable sequence...

Ascii Char code 59 entered.
Ascii Char code 108 entered.
Ascii Char code 115 entered.
Ascii Char code 100 entered.
Ascii Char code 107 entered.
Ascii Char code 102 entered.
Ascii Char code 59 entered.
Ascii Char code 108 entered.
Ascii Char code 115 entered.
Ascii Char code 100 entered.
Ascii Char code 107 entered.
Ascii Char code 102 entered.
Ascii Char code 59 entered.
Ascii Char code 108 entered.
Ascii Char code 115 entered.
Ascii Char code 107 entered.
Ascii Char code 100 entered.
Ascii Char code 59 entered.
Ascii Char code 102 entered.
Ascii Char code 108 entered.
Ascii Char code 115 entered.
Ascii Char code 107 entered.
Ascii Char code 100 entered.

참고 항목

참조

system.Reactive.Subjects 네임스페이스