C# Rxで、1つのストリームを条件によって複数Subscribeする方法

Question

C#でSystem.Reactive.Linqを用いてリアクティブプログラミングを行なっています。
一つのストリームから発行された値を、Subscribeしてある処理を行なっているのですが、その発行された値によって異なる処理を行いたいと考えています。

observable.Subscribe(v => {
                   if(v==1) //処理1
                   else //処理2
});

こんな感じです(スマホで打っているのでインデントがずれていたら申し訳ありません…)。

これを、Subscribeの中で条件分岐するのではなく、このような感じにすることはできないのでしょうか？

observable.SubscribeIf(v=>v==1, v=> //処理1)
                 .subscribe(v=>//処理2);
                 

Whereによるフィルターでは、2回Subscriveを行うことになるのでなんだか…と思っています。
(実際には、もっと多くの条件分岐を行うので尚更)

何かそう言ったメソッド等は用意されていないのでしょうか。
よろしくお願い致します。

Comments

[Zuishin]

反応がなかったら何を求めているのかわからないので他の回答もつかないでしょう。

Answer 1

こんにちは。

Rx では一般的に、処理を分岐させるならその数だけ Subscribe を分けます。
それぞれのパイプラインを独立に処理できるのが Rx を採用するメリットなので、そうしないなら Rx を採用しない方が良いコードになる可能性が高いです。
分岐を扱う場合は、分岐点で Publish を使い明示的にストリームを分岐させるのが良いです。

// パイプラインに分岐点と水栓を作る
var branch = observable.Publish();

// 分岐点からそれぞれ Subscribe
branch.Where(v => v == 1).Subscribe(v => { });
branch.Where(v => v == ??).Subscribe(v => { });

// 水栓を開ける
branch.Connect();

Answer 2

tamoto さんの回答のように Publish を使うのが一般的だと思います。
しかし、あえて自分用に作ってみました。
IObserver<T>, IObservable<T> を実装し、ラムダ式で振る舞いを変えられる汎用クラスである Abyss と、それを使用する拡張メソッド Case です。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;

namespace ConsoleApp1
{
	class Program
	{
		static void Main()
		{
			Observable
				.Range(1, 15)
				.Case(a => a % 15 == 0, _ => Console.WriteLine("FizzBuzz"))
				.Case(a => a % 3 == 0, _ => Console.WriteLine("Fizz"))
				.Case(a => a % 5 == 0, _ => Console.WriteLine("Buzz"))
				.Subscribe(a => Console.WriteLine(a));
		}
	}

	public class Abyss<TUpstream, TDownstream> : IObserver<TUpstream>, IObservable<TDownstream>
	{
		public Abyss(
			IObservable<TUpstream> upstream,
			Action<TUpstream, IEnumerable<IObserver<TDownstream>>> onNext = null,
			Action<Exception, IEnumerable<IObserver<TDownstream>>> onError = null,
			Action<IEnumerable<IObserver<TDownstream>>> onCompleted = null)
		{
			Upstream = upstream;
			OnNextAction = onNext;
			OnErrorAction = onError;
			OnCompletedAction = onCompleted;
		}

		private IDisposable upstreamDisposable = null;

		protected IObservable<TUpstream> Upstream { get; private set; }
		protected Action<TUpstream, IEnumerable<IObserver<TDownstream>>> OnNextAction { get; private set; }
		protected Action<Exception, IEnumerable<IObserver<TDownstream>>> OnErrorAction { get; private set; }
		protected Action<IEnumerable<IObserver<TDownstream>>> OnCompletedAction { get; private set; }

		protected HashSet<IObserver<TDownstream>> Observers { get; }
			= new HashSet<IObserver<TDownstream>>();

		public void OnCompleted()
		{
			DisconnectUpstream();
			if (OnCompletedAction != null)
			{
				OnCompletedAction(Observers.ToArray());
			}
			else
			{
				foreach (var observer in Observers.ToArray())
				{
					observer.OnCompleted();
				}
			}
			Observers.Clear();
		}

		public void OnError(Exception error)
		{
			DisconnectUpstream();
			if (OnErrorAction != null)
			{
				OnErrorAction(error, Observers.ToArray());
			}
			else
			{
				foreach (var observer in Observers.ToArray())
				{
					observer.OnError(error);
				}
			}
			Observers.Clear();
		}

		[System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "<保留中>")]
		public void OnNext(TUpstream value)
		{
			try
			{
				OnNextAction?.Invoke(value, Observers.ToArray());
			}
			catch (Exception e)
			{
				OnError(e);
			}
		}

		public IDisposable Subscribe(IObserver<TDownstream> observer)
		{
			if (observer == null) throw new ArgumentNullException(nameof(observer));
			Observers.Add(observer);
			if (upstreamDisposable == null)
			{
				var connectable = Upstream?.Publish();
				upstreamDisposable = connectable?.Subscribe(this);
				connectable?.Connect();
			}
			return new Disposable(() =>
			{
				Observers.Remove(observer);
				if (Observers.Count == 0)
				{
					DisconnectUpstream();
				}
			});
		}

		private void DisconnectUpstream()
		{
			upstreamDisposable?.Dispose();
			upstreamDisposable = null;
		}

		private class Disposable : IDisposable
		{
			public Disposable(Action action)
			{
				disposeAction = action;
			}

			private readonly Action disposeAction;

			public void Dispose()
			{
				disposeAction?.Invoke();
			}
		}
	}

	public static class AbyssExtension
	{
		public static IObservable<TDownstream> Abyss<TUpstream, TDownstream>(
			this IObservable<TUpstream> upstream,
			Action<TUpstream, IEnumerable<IObserver<TDownstream>>> onNext = null,
			Action<Exception, IEnumerable<IObserver<TDownstream>>> onError = null,
			Action<IEnumerable<IObserver<TDownstream>>> onCompleted = null)
		{
			return new Abyss<TUpstream, TDownstream>(upstream, onNext, onError, onCompleted);
		}

		public static IObservable<T> Case<T>(
			this IObservable<T> upstream,
			Func<T, bool> condition,
			Action<T> action)
		{
			return new Abyss<T, T>(upstream, (data, observers) =>
			{
				if (condition?.Invoke(data) ?? false && action != null)
				{
					action.Invoke(data);
				}
				else
				{
					foreach (var observer in observers)
					{
						observer?.OnNext(data);
					}
				}
			});
		}
	}
}