Observable.Create<TSource> メソッド (Func<IObserver<TSource>, IDisposable>)
サブスクライブ メソッドの実装から監視可能なシーケンスを作成します。
Namespace:System.Reactive.Linq
アセンブリ： System.Reactive (System.Reactive.dll)
構文
'Declaration
Public Shared Function Create(Of TSource) ( _
subscribe As Func(Of IObserver(Of TSource), IDisposable) _
) As IObservable(Of TSource)
'Usage
Dim subscribe As Func(Of IObserver(Of TSource), IDisposable)
Dim returnValue As IObservable(Of TSource)
returnValue = Observable.Create(subscribe)
public static IObservable<TSource> Create<TSource>(
Func<IObserver<TSource>, IDisposable> subscribe
)
public:
generic<typename TSource>
static IObservable<TSource>^ Create(
Func<IObserver<TSource>^, IDisposable^>^ subscribe
)
static member Create :
subscribe:Func<IObserver<'TSource>, IDisposable> -> IObservable<'TSource>
JScript does not support generic types and methods.
型パラメーター
- TSource
ソースの種類。
パラメーター
- subscribe
種類: System.Func<IObserver<TSource>、 IDisposable>
結果として得られる監視可能シーケンスのサブスクライブ メソッドの実装。
戻り値
種類: System.IObservable<TSource>
サブスクライブ メソッドの指定された実装を持つ監視可能なシーケンス。
解説
Create 演算子を使用すると、シーケンスの IObservable<T> インターフェイスを完全に実装することなく、独自のカスタム シーケンスを作成できます。 この演算子を使用すると、iObserver T が型である IObserver<T> を受け取り、IDisposable::D ispose() を呼び出してサブスクリプションを取り消すために使用される IDisposable インターフェイスを返すサブスクライブ関数を実装するだけです。 この演算子の使用は、IObservable<T> インターフェイスの手動実装よりも優先されます。
例
この例では、IObservable<チケットを完全に実装せずに、観察可能な一連のチケットが提供される架空のチケット> システムをシミュレートします。 TicketFactory クラスは、TicketSubscribe という名前の独自のサブスクライブ メソッドを実装します。 このメソッドは、 サブスクライブ パラメーターとして Create 演算子に渡されます。 TicketSubscribe は、TicketSubscribe から返された IDisposable インターフェイスで Dispose メソッドを呼び出してサブスクリプションが取り消されるまで、別のスレッドでチケットの連続シーケンスを作成します。 IObserver<チケット> が TicketSubscribe に渡されます。 シーケンス内の各チケットは、IObserver<チケット>を呼び出すことによって配信されます。OnNext()。 サブスクリプションに対して実行されたオブザーバー アクションには、コンソール ウィンドウに各チケットが表示されます。
using System;
using System.Threading;
using System.Threading.Tasks;
using System.Reactive.Linq;
namespace Example
{
class Program
{
static void Main()
{
IObservable<Ticket> ticketObservable = Observable.Create((Func<IObserver<Ticket>, IDisposable>)TicketFactory.TicketSubscribe);
//*********************************************************************************//
//*** This is a sequence of tickets. Display each ticket in the console window. ***//
//*********************************************************************************//
using(IDisposable handle = ticketObservable.Subscribe(ticket => Console.WriteLine(ticket.ToString())))
{
Console.WriteLine("\nPress ENTER to unsubscribe...\n");
Console.ReadLine();
}
}
}
//***************************************************************************************************//
//*** ***//
//*** The Ticket class just represents a hypothetical ticket composed of an ID and a timestamp. ***//
//*** ***//
//***************************************************************************************************//
class Ticket
{
private readonly string ticketID;
private readonly DateTime timeStamp;
public Ticket(string tid)
{
ticketID = tid;
timeStamp = DateTime.Now;
}
public override string ToString()
{
return String.Format("Ticket ID : {0}\nTimestamp : {1}\n", ticketID, timeStamp.ToString());
}
}
//***************************************************************************************************//
//*** ***//
//*** The TicketFactory class generates a new sequence of tickets on a separate thread. The ***//
//*** generation of the sequence is initiated by the TicketSubscribe method which will be passed ***//
//*** to Observable.Create(). ***//
//*** ***//
//*** TicketSubscribe() provides the IDisposable interface used to dispose of the subscription ***//
//*** stopping ticket generation resources. ***//
//*** ***//
//***************************************************************************************************//
public class TicketFactory : IDisposable
{
private bool bGenerate = true;
internal TicketFactory(object ticketObserver)
{
//************************************************************************//
//*** The sequence generator for tickets will be run on another thread ***//
//************************************************************************//
Task.Factory.StartNew(new Action<object>(TicketGenerator), ticketObserver);
}
//**************************************************************************************************//
//*** Dispose frees the ticket generating resources by allowing the TicketGenerator to complete. ***//
//**************************************************************************************************//
public void Dispose()
{
bGenerate = false;
}
//*****************************************************************************************************************//
//*** TicketGenerator generates a new ticket every 3 sec and calls the observer's OnNext handler to deliver it. ***//
//*****************************************************************************************************************//
private void TicketGenerator(object observer)
{
IObserver<Ticket> ticketObserver = (IObserver<Ticket>)observer;
//***********************************************************************************************//
//*** Generate a new ticket every 3 sec and call the observer's OnNext handler to deliver it. ***//
//***********************************************************************************************//
Ticket t;
while (bGenerate)
{
t = new Ticket(Guid.NewGuid().ToString());
ticketObserver.OnNext(t);
Thread.Sleep(3000);
}
}
//********************************************************************************************************************************//
//*** TicketSubscribe starts the flow of tickets for the ticket sequence when a subscription is created. It is passed to ***//
//*** Observable.Create() as the subscribe parameter. Observable.Create() returns the IObservable<Ticket> that is used to ***//
//*** create subscriptions by calling the IObservable<Ticket>.Subscribe() method. ***//
//*** ***//
//*** The IDisposable interface returned by TicketSubscribe is returned from the IObservable<Ticket>.Subscribe() call. Calling ***//
//*** Dispose cancels the subscription freeing ticket generating resources. ***//
//********************************************************************************************************************************//
public static IDisposable TicketSubscribe(object ticketObserver)
{
TicketFactory tf = new TicketFactory(ticketObserver);
return tf;
}
}
}
コード例からの出力例を次に示します。 Enter キーを押して登録を取り消します。
Press ENTER to unsubscribe...
Ticket ID : a5715731-b9ba-4992-af00-d5030956cfc4
Timestamp : 5/18/2011 6:48:50 AM
Ticket ID : d9797b2b-a356-4928-bfce-797a1637b11d
Timestamp : 5/18/2011 6:48:53 AM
Ticket ID : bb01dc7d-1ed5-4ba0-9ce0-6029187792be
Timestamp : 5/18/2011 6:48:56 AM
Ticket ID : 0d3c95de-392f-4ed3-bbda-fed2c6bc7287
Timestamp : 5/18/2011 6:48:59 AM
Ticket ID : 4d19f79e-6d4f-4fec-83a8-9644a1d4e759
Timestamp : 5/18/2011 6:49:05 AM