2

SubscribeOnブロックを防ぐために別のスレッドを使用するRx サブスクリプションがあります。ただし、リソース管理の問題により、そのサブスクリプションの破棄をブロックしたいと考えています。コンソール アプリまたは winforms アプリのいずれかのコンテキスト内でこれを達成する方法を理解できませんでした (両方のユース ケースがあります)。以下は、私がやっていることをシミュレートする縮小ケースの作業コードです:

internal class Program
{

    private static void Log(string msg)
    {
        Console.WriteLine("[{0}] " + msg, Thread.CurrentThread.ManagedThreadId.ToString());
    }

    private static void Main(string[] args)
    {

        var foo = Observable.Create<long>(obs =>
            {
                Log("Subscribing starting.. this will take a few seconds..");
                Thread.Sleep(TimeSpan.FromSeconds(2));
                var sub =
                    Observable.Interval(TimeSpan.FromSeconds(1))
                              .Do(_ => Log("I am polling..."))
                              .Subscribe(obs);
                return Disposable.Create(() =>
                    {
                        Thread.Sleep(TimeSpan.FromSeconds(3));
                        sub.Dispose();
                        Log("Disposing is really done now!");
                    });
            });

        Log("I am subscribing..");
        var disp = foo.SubscribeOn(NewThreadScheduler.Default).Subscribe(i => Log("Processing " + i.ToString()));
        Log("I have returned from subscribing...");

        // SC.Current is null in a ConsoleApp :/  Can I get a SC that uses my current thread?
        //var dispSynced = new ContextDisposable(SynchronizationContext.Current, disp);
        Thread.Sleep(TimeSpan.FromSeconds(5));
        Log("I'm going to dispose...");
        //dispSynced.Dispose();
        disp.Dispose();
        Log("Disposed has returned...");
        Console.ReadKey();
    }
}

上記を実行すると、次のようになります。

[10] I am subscribing..
[10] I have returned from subscribing...
[11] Subscribing starting.. this will take a few seconds..
[6] I am polling...
[6] Processing 0
[6] I am polling...
[6] Processing 1
[10] I'm going to dispose...
[10] Disposed has returned...
[13] I am polling...
[6] I am polling...
[13] I am polling...
[14] Disposing is really done now!

したがって、私がやろうとしている[10] Disposed has returned...のは、Dispose 呼び出しがブロックされていることを示す最後の行を出力することだけです。

Rx に同梱されている は私の使用例には理想的ですが、現在のスレッドを表すContextDisposableを取得する方法がわかりません。私がやりたいことを行うためにSynchronizationContext使用できる方法はありますか、それともまったく異なるアプローチが必要ですか?ContextDisposable

4

2 に答える 2

2

If you look at the source code for SubscribeOn it looks like the dispose function will be scheduled on the specified scheduler. Try something like this:

private static IObservable<long> GetObservable(IScheduler scheduler)
{
  return Observable.Create<long>(obs =>
  {
    var disposables = new CompositeDisposable();

    disposables.Add(
      Disposable.Create(() =>
      {
        Thread.Sleep(TimeSpan.FromSeconds(3));
        Log("Disposing is really done now!");
      }));

    disposables.Add(
      scheduler.Schedule(() =>
      {
        Log("Subscribing starting.. this will take a few seconds..");
        Thread.Sleep(TimeSpan.FromSeconds(2));
        disposables.Add(
          Observable.Interval(TimeSpan.FromSeconds(1)).Do(_ => Log("I am polling...")).Subscribe(obs));
      }));

    return disposables;
  });


private static void Main(string[] args)
{
  var foo = GetObservable(NewThreadScheduler.Default);

  Log("I am subscribing..");
  var disp = foo.Subscribe(i => Log("Processing " + i.ToString()));
  Log("I have returned from subscribing...");

  // SC.Current is null in a ConsoleApp :/  Can I get a SC that uses my current thread?
  //var dispSynced = new ContextDisposable(SynchronizationContext.Current, disp);
  Thread.Sleep(TimeSpan.FromSeconds(5));
  Log("I'm going to dispose...");
  //dispSynced.Dispose();
  disp.Dispose();
  Log("Disposed has returned...");
  Console.ReadKey();
}
于 2013-01-04T14:59:10.293 に答える
0

（ゼロから始める - 別の方法があります！）

これは調査するのに興味深いものでした-正直なところ、私はこれまでこれを必要としたことはありませんでしたが、観察可能な拡張機能がありObservable.Synchronizeます:

これにより、ブロッキングは非常に簡単になりますが、これがあなたのユースケースに当てはまるかどうかは100％確信が持てません...とにかく、このアプローチを使用して変更されたメインは次のとおりです。

private static void Main(string[] args)
{
    var sync = new object();
    var foo = Observable.CreateWithDisposable<long>(obs =>
    {
        Log("Subscribing starting.. this will take a few seconds..");
        Thread.Sleep(TimeSpan.FromSeconds(2));
        var sub = Observable.Interval(TimeSpan.FromSeconds(1))
                        .Do(_ => Log("I am polling..."))
                        .Synchronize(sync)
                        .Subscribe(obs);
        return Disposable.Create(
            () =>
                {
                    lock (sync)
                    {
                        Thread.Sleep(TimeSpan.FromSeconds(3));
                        sub.Dispose();
                        Log("Disposing is really done now!");                                
                    }
                });
    });

    Log("I am subscribing..");
    var disp = foo
        .SubscribeOn(Scheduler.NewThread)
        .Subscribe(i => Log("Processing " + i));

    Log("I have returned from subscribing...");
    Thread.Sleep(TimeSpan.FromSeconds(5));
    Log("I'm going to dispose...");
    disp.Dispose();
    Log("Disposed has returned...");
    Console.ReadKey();
}
于 2013-01-03T00:25:58.170 に答える