C#でのリアクティブプログラミング入門：実践ガイド

Install-Package System.Reactive

using System.Reactive.Linq;
using System.Reactive.Subjects;

var observable = Observable.Range(1, 10);
observable.Subscribe(x => Console.WriteLine(x));

var evenNumbers = observable.Where(x => x % 2 == 0);
evenNumbers.Subscribe(x => Console.WriteLine($"Even number: {x}"));

var subject = new Subject<int>();
subject.Subscribe(x => Console.WriteLine($"Subscriber 1: {x}"));
subject.Subscribe(x => Console.WriteLine($"Subscriber 2: {x}"));
subject.OnNext(1);
subject.OnNext(2);

var interval = Observable.Interval(TimeSpan.FromSeconds(1));
interval.Subscribe(x => Console.WriteLine($"Tick: {x}"));

var observable = Observable.Create<int>(observer =>
{
    for (int i = 0; i < 5; i++)
    {
        observer.OnNext(i);
    }
    observer.OnCompleted();
    return Disposable.Empty;
});
observable.Subscribe(x => Console.WriteLine($"Received: {x}"));

var observer = Observer.Create<int>(
    onNext: x => Console.WriteLine($"Next: {x}"),
    onError: ex => Console.WriteLine($"Error: {ex.Message}"),
    onCompleted: () => Console.WriteLine("Completed")
);
observable.Subscribe(observer);

var observable = Observable.Interval(TimeSpan.FromSeconds(1));
var observer = Observer.Create<long>(
    onNext: x => Console.WriteLine($"Tick: {x}"),
    onError: ex => Console.WriteLine($"Error: {ex.Message}"),
    onCompleted: () => Console.WriteLine("Completed")
);
observable.Subscribe(observer);

var sensorData = Observable.Interval(TimeSpan.FromSeconds(1))
                           .Select(x => new { Time = DateTime.Now, Value = x });

sensorData.Subscribe(data => Console.WriteLine($"Time: {data.Time}, Value: {data.Value}"));

var buttonClicks = Observable.FromEventPattern<EventArgs>(button, "Click");

buttonClicks.Subscribe(_ => Console.WriteLine("Button clicked!"));

var client = new HttpClient();
var apiResponse = Observable.FromAsync(() => client.GetStringAsync("https://api.example.com/data"));

apiResponse.Subscribe(response => Console.WriteLine($"API Response: {response}"));

var numbers = Observable.Range(1, 10);
var evenNumbers = numbers.Where(x => x % 2 == 0);

evenNumbers.Subscribe(x => Console.WriteLine($"Even number: {x}"));

var observable = Observable.Create<int>(observer =>
{
    observer.OnNext(1);
    observer.OnNext(2);
    observer.OnError(new Exception("An error occurred!"));
    observer.OnNext(3);
    return Disposable.Empty;
});

observable.Subscribe(
    x => Console.WriteLine($"Received: {x}"),
    ex => Console.WriteLine($"Error: {ex.Message}"),
    () => Console.WriteLine("Completed")
);

var retryObservable = observable.Retry(3);

retryObservable.Subscribe(
    x => Console.WriteLine($"Received: {x}"),
    ex => Console.WriteLine($"Error after retries: {ex.Message}"),
    () => Console.WriteLine("Completed")
);

var fallbackObservable = Observable.Return(100);
var catchObservable = observable.Catch(fallbackObservable);

catchObservable.Subscribe(
    x => Console.WriteLine($"Received: {x}"),
    ex => Console.WriteLine($"Error: {ex.Message}"),
    () => Console.WriteLine("Completed")
);

var finallyObservable = observable.Finally(() => Console.WriteLine("Stream ended."));

finallyObservable.Subscribe(
    x => Console.WriteLine($"Received: {x}"),
    ex => Console.WriteLine($"Error: {ex.Message}"),
    () => Console.WriteLine("Completed")
);

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);
observable.ObserveOn(uiScheduler)
          .Subscribe(x => Console.WriteLine($"UI Thread: {x}"));

observable.SubscribeOn(ThreadPoolScheduler.Instance)
          .Subscribe(x => Console.WriteLine($"ThreadPool: {x}"));

observable.SubscribeOn(ThreadPoolScheduler.Instance)
          .ObserveOn(uiScheduler)
          .Subscribe(x => Console.WriteLine($"ThreadPool to UI Thread: {x}"));

public class CustomScheduler : IScheduler
{
    public DateTimeOffset Now => DateTimeOffset.Now;

    public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action)
    {
        return new ScheduledItem<TState>(this, state, action).Invoke();
    }

    public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action)
    {
        var delay = dueTime - Now;
        return new ScheduledItem<TState>(this, state, action, delay).Invoke();
    }
}

public partial class MainWindow : Window
{
    public MainWindow()
    {
        InitializeComponent();

        var buttonClicks = Observable.FromEventPattern<RoutedEventArgs>(myButton, "Click");
        buttonClicks.Subscribe(_ => myTextBox.Text = "Button clicked!");
    }
}

public MainWindow()
{
    InitializeComponent();

    var textChanges = Observable.FromEventPattern<TextChangedEventArgs>(myTextBox, "TextChanged")
                                .Select(evt => ((TextBox)evt.Sender).Text);

    textChanges.Throttle(TimeSpan.FromMilliseconds(500))
               .DistinctUntilChanged()
               .ObserveOnDispatcher()
               .Subscribe(text => myLabel.Content = $"You typed: {text}");
}

public partial class MainPage : ContentPage
{
    public MainPage()
    {
        InitializeComponent();

        var buttonClicks = Observable.FromEventPattern<EventArgs>(myButton, "Clicked");
        buttonClicks.Subscribe(_ => myLabel.Text = "Button clicked!");
    }
}

public MainPage()
{
    InitializeComponent();

    var textChanges = Observable.FromEventPattern<TextChangedEventArgs>(myEntry, "TextChanged")
                                .Select(evt => ((Entry)evt.Sender).Text);

    textChanges.Throttle(TimeSpan.FromMilliseconds(500))
               .DistinctUntilChanged()
               .ObserveOn(SynchronizationContext.Current)
               .Subscribe(text => myLabel.Text = $"You typed: {text}");
}

var numbers = Observable.Range(1, 10);
numbers.Subscribe(x => 
{
    if (x % 2 == 0)
        Console.WriteLine($"Even number: {x}");
    else
        Console.WriteLine($"Odd number: {x}");
});

var observable = Observable.Create<int>(observer =>
{
    for (int i = 1; i <= 5; i++)
    {
        if (i == 3)
        {
            observer.OnError(new Exception("An error occurred at value 3!"));
            return Disposable.Empty;
        }
        observer.OnNext(i);
    }
    observer.OnCompleted();
    return Disposable.Empty;
});

observable.Subscribe(
    x => Console.WriteLine($"Received: {x}"),
    ex => Console.WriteLine($"Error: {ex.Message}"),
    () => Console.WriteLine("Completed")
);

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);
var interval = Observable.Interval(TimeSpan.FromSeconds(1));

interval.ObserveOn(uiScheduler)
        .Subscribe(x => Console.WriteLine($"UI Thread: {x}"));

public partial class MainWindow : Window
{
    public MainWindow()
    {
        InitializeComponent();

        var textChanges = Observable.FromEventPattern<TextChangedEventArgs>(myTextBox, "TextChanged")
                                    .Select(evt => ((TextBox)evt.Sender).Text);

        textChanges.Throttle(TimeSpan.FromMilliseconds(500))
                   .DistinctUntilChanged()
                   .ObserveOnDispatcher()
                   .Subscribe(text => myLabel.Content = $"You typed: {text}");
    }
}