Reactive Extension's Where method reimplemented

reactive-where.cs

public static class ObservableExtensions
{
	public static IObservable<TSource> Where<TSource> (this IObservable<TSource> source, Func<TSource,bool> predicate)
	{
		if (source == null)
		{
			throw new ArgumentNullException("source");
		}
		if (predicate == null)
		{
			throw new ArgumentNullException("predicate");
		}
	
		// consider whether this should be wrapped in Observable.Defer( () => {} )  
		return Observable.Create<TSource>(observer =>
			{
				return source.Subscribe(x=>
					{
						bool flag = false;
						try
						{
							flag = predicate(x);
						}
						catch (Exception e)
						{
							observer.OnError(e);
							return;
						}
						if (flag)
						{
							observer.OnNext(x);
						}
					},
					observer.OnError,
					observer.OnCompleted
					);
			});
	}
}