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> 型は、次のメンバーを公開します。

メソッド

	名前	形容
	OnCompleted	(IObserver<TSource>から継承されます)。
	OnError	(IObserver<TSource>から継承されます)。
	OnNext	(IObserver<TSource>から継承されます)。
	サブスクライブ	(IObservable<TResult>から継承されます)。

ページのトップへ

拡張メソッド

ページのトップへ

備考

このサブジェクト インターフェイスは、ある種類の観測可能なシーケンスを観察しながら、別の型の監視可能なシーケンスを公開する際に、対象に柔軟性を提供します。

例

この例では、ISubject<TSource、TResult> を実装し、別の型の監視可能なシーケンスを発行しながら、1 つの型の監視可能なシーケンスを観察する方法を示します。 この例の 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 名前空間の