MinInt32
namespace System.
Reactive.
Linq.
ObservableImpl
{
internal sealed class MinInt32 :
Producer<
int,
MinInt32.
_>
{
internal sealed class _ :
Sink<
int>,
IObserver<
int>
{
private bool _hasValue;
private int _lastValue;
public _(
IObserver<
int>
observer,
IDisposable cancel)
:
base(
observer,
cancel)
{
_hasValue =
false;
_lastValue =
0;
}
public void OnNext(
int value)
{
if (
_hasValue) {
if (
value <
_lastValue)
_lastValue =
value;
}
else {
_lastValue =
value;
_hasValue =
true;
}
}
public void OnError(
Exception error)
{
_observer.
OnError(
error);
base.
Dispose();
}
public void OnCompleted()
{
if (!
_hasValue)
_observer.
OnError(
new InvalidOperationException(
Strings_Linq.
NO_ELEMENTS));
else {
_observer.
OnNext(
_lastValue);
_observer.
OnCompleted();
}
base.
Dispose();
}
}
private readonly IObservable<
int>
_source;
public MinInt32(
IObservable<
int>
source)
{
_source =
source;
}
protected override _ CreateSink(
IObserver<
int>
observer,
IDisposable cancel)
{
return new _(
observer,
cancel);
}
protected override IDisposable Run(
_ sink)
{
return _source.
SubscribeSafe(
sink);
}
}
}