Observable.Create<TSource> 方法 (Func<IObserver<TSource>、IDisposable>)

从订阅方法实现创建可观察序列。

Namespace:System.Reactive.Linq
装配: System.Reactive.dll) 中的 System.Reactive (

语法

'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
    源的类型。

parameters

返回值

类型: System.IObservable<TSource>
具有订阅方法的指定实现的可观测序列。

备注

Create 运算符允许你创建自己的自定义序列,而无需为序列完全实现 IObservable<T> 接口。 使用此运算符,只需实现一个订阅函数,该函数采用 IObserver<T> ,其中 T 是类型,并返回 IDisposable 接口,该接口用于通过调用 IDisposable::D ispose () 取消订阅。 使用此运算符优先于 IObservable<T> 接口的手动实现。

示例

此示例模拟一个假设的票证系统,其中提供了可观测的票证序列,但未完全实现 IObservable<Ticket>。 TicketFactory 类实现其自己的订阅方法,称为 TicketSubscribe。 此方法作为 subscribe 参数传递给 Create 运算符。 TicketSubscribe 通过调用从 TicketSubscribe 返回的 IDisposable 接口上的 Dispose 方法,在另一个线程上创建一个连续的票证序列,直到取消订阅。 IObserver<票证> 传递到 TicketSubscribe。 序列中的每个票证都是通过调用 IObserver Ticket> 传递的<。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

另请参阅

参考

可观测类

创建重载

System.Reactive.Linq 命名空间