### Simplified Subscription Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/02_KeyTypes.md

An example demonstrating the common usage pattern of subscribing to an observable sequence using a lambda expression for the onNext action. This leverages the Subscribe extension method for brevity.

```csharp
ticks.Subscribe(
    tick => Console.WriteLine($"Tick {tick}"));
```

--------------------------------

### Running Scheduled Actions with TestScheduler.Start()

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Use Start() to execute all scheduled actions. The virtual clock advances to the time of the last scheduled item. Ensure all necessary actions are scheduled before calling Start().

```csharp
var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A")); // Schedule immediately
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));

Console.WriteLine("scheduler.Start();");
scheduler.Start();

Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);
```

--------------------------------

### Example of repeating an Observable sequence

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/09_CombiningSequences.md

This example demonstrates repeating the sequence [0,1,2] three times. The output shows the sequence repeated, followed by the completion message.

```csharp
var source = Observable.Range(0, 3);
var result = source.Repeat(3);

result.Subscribe(
    Console.WriteLine,
    () => Console.WriteLine("Completed"));
```

--------------------------------

### TestScheduler.Start Method Signature

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This overload of the Start method allows precise control over the timing of observable creation, subscription, and disposal. Time is measured in ticks.

```csharp
public ITestableObserver<T> Start<T>(
    Func<IObservable<T>> create, 
    long created, 
    long subscribed, 
    long disposed)
{...}
```

--------------------------------

### Simple Recursive Scheduling Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/11_SchedulingAndThreading.md

Demonstrates the simplest recursive overload of Schedule. The provided action recursively calls itself, and Rx handles the looping and cancellation.

```csharp
Action<Action> work = (Action self) =>
{
    Console.WriteLine("Running");
    self();
};

var token = s.Schedule(work);
    
Console.ReadLine();
Console.WriteLine("Cancelling");

token.Dispose();

Console.WriteLine("Cancelled");
```

--------------------------------

### Illustrative Rx.NET ObserveOn Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/adr/0005-package-split.md

This example demonstrates a scenario where ObserveOn is used with a SynchronizationContext, which can lead to compiler errors in specific project configurations due to transitive framework references.

```cs
using System.Reactive.Linq;

SynchronizationContext.SetSynchronizationContext(new SynchronizationContext());

IObservable<int> numbers = Observable.Range(1, 10);
IObservable<int> numbersViaSyncContext = numbers.ObserveOn(SynchronizationContext.Current!);
numbers.Subscribe(x => Console.WriteLine($"Number: {x}"));
```

--------------------------------

### TestScheduler.Start Output Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This is the expected output when testing Observable.Interval with immediate subscription and disposal after 5 seconds. It shows the virtual time, the number of notifications, and each recorded message with its timestamp.

```text
Time is 50000000 ticks
Received 5 notifications
OnNext(0) @ 10000001
OnNext(1) @ 20000001
OnNext(2) @ 30000001
OnNext(3) @ 40000001
OnCompleted() @ 40000001
```

--------------------------------

### Hot Observable Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Illustrates creating and testing a hot observable. Note that notifications occur at their scheduled times relative to the observable's creation, regardless of subscription time.

```csharp
var scheduler = new TestScheduler();
var source = scheduler.CreateHotObservable(
    new Recorded<Notification<long>>(10000000, Notification.CreateOnNext(0L)),
    new Recorded<Notification<long>>(20000000, Notification.CreateOnNext(1L)),
    new Recorded<Notification<long>>(30000000, Notification.CreateOnNext(2L)),
    new Recorded<Notification<long>>(40000000, Notification.CreateOnNext(3L)),
    new Recorded<Notification<long>>(40000000, Notification.CreateOnCompleted<long>()) 
    );

var testObserver = scheduler.Start(
    () => source,
    0,
    0,
    TimeSpan.FromSeconds(5).Ticks);

Console.WriteLine("Time is {0} ticks", scheduler.Clock);
Console.WriteLine("Received {0} notifications", testObserver.Messages.Count);

foreach (Recorded<Notification<long>> message in testObserver.Messages)
{
    Console.WriteLine("  {0} @ {1}", message.Value, message.Time);
}
```

--------------------------------

### Subscribing to new ships with GroupBy

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/08_Partitioning.md

This example shows how to subscribe to the observable sequence of grouped observables. Each time a new ship (key) is encountered, a message is printed to the console.

```csharp
perShipObservables.Subscribe(m => Console.WriteLine($"New ship! {m.Key}"));
```

--------------------------------

### Testing with Delayed Subscription

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This example shows how delaying the subscription time affects the received notifications. With a subscription at 2 seconds and disposal at 5 seconds, some messages are missed.

```csharp
var testObserver = scheduler.Start(
    () => Observable.Interval(TimeSpan.FromSeconds(1), scheduler).Take(4), 
    0,
    TimeSpan.FromSeconds(2).Ticks,
    TimeSpan.FromSeconds(5).Ticks);
```

--------------------------------

### Buffer with Range and Dump Output

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/08_Partitioning.md

This artificial example demonstrates how Buffer handles incomplete chunks when the source completes. The final chunk may be smaller than the specified size.

```csharp
Observable
    .Range(1, 5)
    .Buffer(2)
    .Select(chunk => string.Join(", ", chunk))
    .Dump("chunks");
```

--------------------------------

### ViewModel Test Setup with TestSchedulers

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Set up a ViewModel for testing by providing a mock model and a TestSchedulers instance. This allows for controlled testing of ViewModel logic.

```csharp
[TestInitialize]
public void SetUp()
{
    _myModelMock = new Mock<IMyModel>();
    _schedulerProvider = new TestSchedulers();
    _viewModel = new MyViewModel(_myModelMock.Object, _schedulerProvider);
}
```

--------------------------------

### AdvanceBy Method Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Shows how to use AdvanceBy to move the virtual clock forward by a relative amount, executing scheduled actions that become due during the advancement. Time is measured in ticks.

```csharp
var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A")); // Schedule immediately
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));

Console.WriteLine("scheduler.AdvanceBy(1);");
scheduler.AdvanceBy(1);

Console.WriteLine("scheduler.AdvanceBy(9);");
scheduler.AdvanceBy(9);

Console.WriteLine("scheduler.AdvanceBy(5);");
scheduler.AdvanceBy(5);

Console.WriteLine("scheduler.AdvanceBy(5);");
scheduler.AdvanceBy(5);
```

--------------------------------

### Implementing Sum with Aggregate

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

This example demonstrates how to implement a Sum operation for integers using Aggregate. The accumulator function adds the current element to the accumulator.

```csharp
Func<int, int, int> s = (acc, element) => acc + element;
IObservable<int> sum = source.Aggregate(s);
```

--------------------------------

### TestScheduler.Start Parameterless Overload

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This is the simplest overload of Start, taking only the observable factory. It uses default values for creation, subscription, and disposal times, making it convenient for basic tests.

```csharp
public ITestableObserver<T> Start<T>(Func<IObservable<T>> create)
{
    if (create == null)
    {
        throw new ArgumentNullException("create");
    }
    else
    {
        return this.Start<T>(create, 100L, 200L, 1000L);
    }
}
```

--------------------------------

### Aggregate Call Example for Count (Second Element)

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Shows the subsequent call to the accumulator function with the updated accumulator value and the next source element.

```csharp
c(1, 200) // returns 2
```

--------------------------------

### Aggregate Call Example for Sum (First Element)

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Illustrates the first call to the Sum accumulator function, using the default initial accumulator value (0) and the first element.

```csharp
s(0, 100) // returns 100
```

--------------------------------

### Calculating Running Count and Sum with Scan

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

This example demonstrates using Scan to compute a running tuple of count and sum for a sequence of numbers. The accumulator is updated with each element from the source.

```csharp
IObservable<int> nums = Observable.Range(1, 5);

IObservable<(int Count, int Sum)> avgAcc = nums.Scan(
    (Count: 0, Sum: 0),
    (acc, element) => (Count: acc.Count + 1, Sum: acc.Sum + element));

avgAcc.Dump("acc");
```

--------------------------------

### Buffer elements by time in Rx.NET

Source: https://context7.com/dotnet/reactive/llms.txt

Collects items into chunks and emits each chunk as an IList<T>. This example buffers by time, collecting items every 500ms.

```csharp
// Buffer by time: collect items every 500ms
IObservable<IList<SensorReading>> batchedReadings = 
    sensorStream.Buffer(TimeSpan.FromMilliseconds(500));
```

--------------------------------

### Testing Observable.Interval with TestScheduler

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This example demonstrates testing Observable.Interval by creating and subscribing to it immediately, then disposing after 5 seconds. It outputs the recorded notifications and the final virtual time.

```csharp
var scheduler = new TestScheduler();
var source = Observable.Interval(TimeSpan.FromSeconds(1), scheduler)
    .Take(4);

var testObserver = scheduler.Start(
    () => source, 
    0, 
    0, 
    TimeSpan.FromSeconds(5).Ticks);

Console.WriteLine("Time is {0} ticks", scheduler.Clock);
Console.WriteLine("Received {0} notifications", testObserver.Messages.Count);

foreach (Recorded<Notification<long>> message in testObserver.Messages)
{
    Console.WriteLine("{0} @ {1}", message.Value, message.Time);
}
```

--------------------------------

### Subscribing to a Custom Observable Sequence

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Demonstrates how to subscribe to a custom IObservable<T> implementation and handle the emitted values and completion notification. This example uses Console.WriteLine for output.

```csharp
var numbers = new MySequenceOfNumbers();
numbers.Subscribe(
    number => Console.WriteLine($"Received value: {number}"),
    () => Console.WriteLine("Sequence terminated"));
```

--------------------------------

### TestScheduler.Start Overload with Disposal Time

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

This overload of Start simplifies testing by allowing only the disposal time to be specified. Default values are used for creation and subscription times.

```csharp
public ITestableObserver<T> Start<T>(Func<IObservable<T>> create, long disposed)
{
    if (create == null)
    {
        throw new ArgumentNullException("create");
    }
    else
    {
        return this.Start<T>(create, 100L, 200L, disposed);
    }
}
```

--------------------------------

### Aggregate Call Example for Sum (Second Element)

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Shows the second call to the Sum accumulator function, using the previously returned accumulator value and the next source element.

```csharp
s(100, 200) // returns 300
```

--------------------------------

### Implementing Buffer with Window Operator

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/08_Partitioning.md

Provides a custom implementation of the Buffer operator using the Window operator. This example shows how to collect items into a List within each window using Aggregate.

```csharp
public static IObservable<IList<T>> MyBuffer<T>(this IObservable<T> source, int count)
{
    return source.Window(count)
        .SelectMany(window => 
            window.Aggregate(
                new List<T>(), 
                (list, item) =>
                {
                    list.Add(item);
                    return list;
                }));
}
```

--------------------------------

### Create windows of observable elements in Rx.NET

Source: https://context7.com/dotnet/reactive/llms.txt

Like Buffer, but emits each chunk as a nested IObservable<T>, enabling streaming processing within each window. This example sums elements within each window.

```csharp
IObservable<IObservable<int>> windows = 
    Observable.Range(1, 10).Window(3);

windows.SelectMany(window => window.Sum())
       .Subscribe(sum => Console.WriteLine($"Window sum: {sum}"));
// Output: Window sum: 6, Window sum: 15, Window sum: 24, Window sum: 10
```

--------------------------------

### Using PublishLast to Replay the Final Value

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

This example demonstrates how PublishLast ensures that even late subscribers receive the final value emitted by the source observable. The Connect() method must be called to start the observable sequence.

```csharp
IConnectableObservable<long> pticks = Observable
    .Interval(TimeSpan.FromSeconds(0.1))
    .Take(4)
    .PublishLast();

pticks.Subscribe(x => Console.WriteLine($"Sub1: {x} ({DateTime.Now})"));
pticks.Subscribe(x => Console.WriteLine($"Sub2: {x} ({DateTime.Now})"));

pticks.Connect();
Thread.Sleep(3000);
Console.WriteLine();
Console.WriteLine("Adding more subscribers");
Console.WriteLine();

pticks.Subscribe(x => Console.WriteLine($"Sub3: {x} ({DateTime.Now})"));
pticks.Subscribe(x => Console.WriteLine($"Sub4: {x} ({DateTime.Now})"));
```

--------------------------------

### Set up Rx.NET Project

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/01_WhyRx.md

Use the .NET CLI to create a new console project and add the System.Reactive NuGet package.

```powershell
mkdir TryRx
cd TryRx
dotnet new console
dotnet add package System.Reactive
```

--------------------------------

### Timed Sequence Generation with Observable.Interval

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Creates an observable sequence that publishes incremental values starting from zero at a specified interval. This example publishes values every 250 milliseconds. Subscriptions must be disposed to stop the infinite sequence.

```csharp
IObservable<long> interval = Observable.Interval(TimeSpan.FromMilliseconds(250));
interval.Subscribe(
    Console.WriteLine,
    () => Console.WriteLine("completed"));
```

--------------------------------

### Observable Creation Without Deferral

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Demonstrates immediate execution of startup work when calling a factory method, even before subscription. Use when startup work should always occur.

```csharp
static IObservable<int> WithoutDeferal()
{
    Console.WriteLine("Doing some startup work...");
    return Observable.Range(1, 3);
}

Console.WriteLine("Calling factory method");
IObservable<int> s = WithoutDeferal();

Console.WriteLine("First subscription");
s.Subscribe(Console.WriteLine);

Console.WriteLine("Second subscription");
s.Subscribe(Console.WriteLine);
```

--------------------------------

### Using Observable.Start with Action in Rx.NET

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Demonstrates converting an Action delegate to an IObservable<Unit> using Observable.Start. The observable emits Unit upon completion and signals completion.

```csharp
static void StartAction()
{
    var start = Observable.Start(() =>
        {
            Console.Write("Working away");
            for (int i = 0; i < 10; i++)
            {
                Thread.Sleep(100);
                Console.Write(".");
            }
        });

    start.Subscribe(
        unit => Console.WriteLine("Unit published"), 
        () => Console.WriteLine("Action completed"));
}
```

--------------------------------

### Scheduling Actions After Start() with TestScheduler

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Actions scheduled after Start() has completed will not execute until Start() is called again. The virtual clock does not automatically advance for newly scheduled items.

```csharp
var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A"));
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));

Console.WriteLine("scheduler.Start();");
scheduler.Start();

Console.WriteLine("scheduler.Clock:{0}", scheduler.Clock);

scheduler.Schedule(() => Console.WriteLine("D"));
```

--------------------------------

### Clone and Set Up Remote

Source: https://github.com/dotnet/reactive/wiki/Contributing-Code

Clone the project repository and add a remote reference to your personal fork. Replace 'user' with your GitHub username.

```bash
git clone https://github.com/dotnet/reactive.git

cd reactive

git remote add myfork https://github.com/user/reactive.git
```

--------------------------------

### AdvanceTo Method Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/16_TestingRx.md

Demonstrates scheduling actions at different times and using AdvanceTo to move the virtual clock to specific points, executing scheduled actions as the clock reaches them. Actions scheduled before the target time are executed.

```csharp
var scheduler = new TestScheduler();
scheduler.Schedule(() => Console.WriteLine("A")); // Schedule immediately
scheduler.Schedule(TimeSpan.FromTicks(10), () => Console.WriteLine("B"));
scheduler.Schedule(TimeSpan.FromTicks(20), () => Console.WriteLine("C"));

Console.WriteLine("scheduler.AdvanceTo(1);");
scheduler.AdvanceTo(1);

Console.WriteLine("scheduler.AdvanceTo(10);");
scheduler.AdvanceTo(10);

Console.WriteLine("scheduler.AdvanceTo(15);");
scheduler.AdvanceTo(15);

Console.WriteLine("scheduler.AdvanceTo(20);");
scheduler.AdvanceTo(20);
```

--------------------------------

### Observable Interval with NewThreadScheduler

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/11_SchedulingAndThreading.md

This example demonstrates a source that deliberately causes each notification to arrive on a different thread by using Observable.Return with NewThreadScheduler for each tick. This is an illustrative example of an unruly source.

```csharp
Observable
    .Interval(TimeSpan.FromSeconds(1))
    .SelectMany(tick => Observable.Return(tick, NewThreadScheduler.Default))
    .Subscribe(tick => 
      Console.WriteLine($"{DateTime.Now}-{Environment.CurrentManagedThreadId}: Tick {tick}"));
```

--------------------------------

### Observable Creation With Deferral

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Illustrates deferred execution of startup work until the moment of subscription using Observable.Defer. This ensures work is only done when needed.

```csharp
static IObservable<int> WithDeferal()
{
    return Observable.Defer(() =>
    {
        Console.WriteLine("Doing some startup work...");
        return Observable.Range(1, 3);
    });
}
```

--------------------------------

### Using Observable.Start with Func<T> in Rx.NET

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Shows how to convert a Func<T> delegate returning a value into an IObservable<T> using Observable.Start. The observable emits the function's return value and then completes.

```csharp
static void StartFunc()
{
    var start = Observable.Start(() =>
    {
        Console.Write("Working away");
        for (int i = 0; i < 10; i++)
        {
            Thread.Sleep(100);
            Console.Write(".");
        }
        return "Published value";
    });

    start.Subscribe(
        Console.WriteLine, 
        () => Console.WriteLine("Action completed"));
}
```

--------------------------------

### Get First or Default Moving Vessel Navigation Event with Timeout

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/05_Filtering.md

Filters messages for a specific MMSI, casts them to IVesselNavigation, applies a timeout using TakeUntil, and then attempts to get the first event indicating movement. If no movement is detected within 5 minutes, it emits null.

```csharp
IObservable<IVesselNavigation?> moving = 
   receiverHost.Messages
    .Where(v => v.Mmsi == exampleMmsi)
    .OfType<IVesselNavigation>()
    .TakeUntil(DateTimeOffset.Now.AddMinutes(5))
    .FirstOrDefaultAsync(vn => vn.SpeedOverGround > 1f);
```

--------------------------------

### Timeout with DateTimeOffset Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/12_Timing.md

Illustrates using Timeout with a DateTimeOffset. The sequence will error with a TimeoutException if it does not complete by the specified absolute time.

```csharp
var dueDate = DateTimeOffset.UtcNow.AddSeconds(4);
var source = Observable.Interval(TimeSpan.FromSeconds(1));
var timeout = source.Timeout(dueDate);
timeout.Subscribe(
    Console.WriteLine,
    Console.WriteLine,
    () => Console.WriteLine("Completed"));
```

--------------------------------

### Observer Invocation on Same Thread

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/11_SchedulingAndThreading.md

This example shows that observers are invoked on the same thread that calls OnNext. This behavior is default and efficient for simple scenarios.

```csharp
object sync = new();
ParameterizedThreadStart notify = arg =>
{
    string message = arg?.ToString() ?? "null";
    Console.WriteLine(
        $"OnNext({message}) on thread: {Environment.CurrentManagedThreadId}");
    lock (sync)
    {
        subject.OnNext(message);
    }
};

notify("Main");
new Thread(notify).Start("First worker thread");
new Thread(notify).Start("Second worker thread");
```

--------------------------------

### Conceptual Representation of Count with Aggregate

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Illustrates the nested application of the accumulator function for recreating the Count operation, starting with an initial accumulator of 0.

```csharp
int sum = s(s(s(s(s(0, 1), 2), 3), 4), 5);
// Note: Aggregate doesn't return this directly -
// it returns an IObservable<int> that produces this value.
```

--------------------------------

### Conceptual Representation of Aggregate

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

This shows the nested application of the accumulator function for a source producing values 1 through 5, starting with a default accumulator.

```csharp
T result = f(f(f(f(f(default(T), 1), 2), 3), 4), 5);
```

--------------------------------

### Buffer elements by count in Rx.NET

Source: https://context7.com/dotnet/reactive/llms.txt

Collects items into chunks and emits each chunk as an IList<T>. This example buffers by count, emitting groups of 3.

```csharp
// Buffer by count: emit groups of 3
Observable.Range(1, 10)
    .Buffer(3)
    .Subscribe(batch => Console.WriteLine(string.Join(", ", batch)));
// Output: [1,2,3], [4,5,6], [7,8,9], [10]
```

--------------------------------

### Oversimplified ToObservable for IEnumerable<T>

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

A basic, oversimplified implementation of converting an IEnumerable<T> to an IObservable<T> using Observable.Create. This example is for illustration and does not handle unsubscription correctly.

```csharp
// Example code only - do not use!
public static IObservable<T> ToObservableOversimplified<T>(this IEnumerable<T> source)
{
    return Observable.Create<T>(o =>
    {
        foreach (var item in source)
        {
            o.OnNext(item);
        }

        o.OnCompleted();

        // Incorrectly ignoring unsubscription.
        return Disposable.Empty;
    });
}
```

--------------------------------

### Demonstrating Publish with Late Subscribers

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

Shows how subscribers added after Connect only receive events that occur after their subscription. Requires Rx.NET.

```csharp
IConnectableObservable<long> publishedTicks = Observable
    .Interval(TimeSpan.FromSeconds(1))
    .Take(4)
    .Publish();

publishedTicks.Subscribe(x => Console.WriteLine($"Sub1: {x} ({DateTime.Now})"));
publishedTicks.Subscribe(x => Console.WriteLine($"Sub2: {x} ({DateTime.Now})"));

publishedTicks.Connect();
Thread.Sleep(2500);
Console.WriteLine();
Console.WriteLine("Adding more subscribers");
Console.WriteLine();

publishedTicks.Subscribe(x => Console.WriteLine($"Sub3: {x} ({DateTime.Now})"));
publishedTicks.Subscribe(x => Console.WriteLine($"Sub4: {x} ({DateTime.Now})"));
```

--------------------------------

### Replay Operator Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

Demonstrates the Replay operator's behavior where late subscribers receive previously emitted events. Requires Rx.NET.

```csharp
IConnectableObservable<long> pticks = Observable
    .Interval(TimeSpan.FromSeconds(1))
    .Take(4)
    .Replay();

pticks.Subscribe(x => Console.WriteLine($"Sub1: {x} ({DateTime.Now})"));
pticks.Subscribe(x => Console.WriteLine($"Sub2: {x} ({DateTime.Now})"));

pticks.Connect();
Thread.Sleep(2500);
Console.WriteLine();
Console.WriteLine("Adding more subscribers");
Console.WriteLine();

pticks.Subscribe(x => Console.WriteLine($"Sub3: {x} ({DateTime.Now})"));
pticks.Subscribe(x => Console.WriteLine($"Sub4: {x} ({DateTime.Now})"));
```

--------------------------------

### Timeout with TimeSpan Example

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/12_Timing.md

Demonstrates using Timeout with a TimeSpan. If the source observable does not produce a value within the specified TimeSpan, a TimeoutException is thrown.

```csharp
var source = Observable.Interval(TimeSpan.FromMilliseconds(100))
                       .Take(5)
                       .Concat(Observable.Interval(TimeSpan.FromSeconds(2)));

var timeout = source.Timeout(TimeSpan.FromSeconds(1));
timeout.Subscribe(
    Console.WriteLine,
    Console.WriteLine,
    () => Console.WriteLine("Completed"));
```

--------------------------------

### Implementing Aggregate with Scan and TakeLast

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Shows how the Aggregate operator can be simulated using Scan followed by TakeLast. This illustrates the relationship between the two operators.

```csharp
source.Aggregate(0, (acc, current) => acc + current);
// is equivalent to 
source.Scan(0, (acc, current) => acc + current).TakeLast();
```

--------------------------------

### Using Multicast with a Subject

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

Demonstrates the direct use of the Multicast operator, requiring an explicit Subject to be provided. Multiple subscribers are set up before Connect is called.

```csharp
IConnectableObservable<int> m = src.Multicast(new Subject<int>());

m.Subscribe(x => Console.WriteLine($"Sub1: {x}"));
m.Subscribe(x => Console.WriteLine($"Sub2: {x}"));
m.Subscribe(x => Console.WriteLine($"Sub3: {x}"));

m.Connect();
```

--------------------------------

### Aggregate Call Example for Count

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/07_Aggregation.md

Illustrates how the Aggregate method calls the accumulator function with the current accumulator value (initially 0) and the source element.

```csharp
c(0, 100) // returns 1
```

--------------------------------

### Wrap FileSystemWatcher with Rx

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

This C# code demonstrates how to wrap the .NET FileSystemWatcher to expose filesystem events as an IObservable<FileSystemEventArgs>. It includes basic error handling and synchronization for observer calls. Note that this is a simplified example and may not be efficient for multiple subscribers or handle errors optimally.

```csharp
public class RxFsEvents : IObservable<FileSystemEventArgs>
{
    private readonly string folder;

    public RxFsEvents(string folder)
    {
        this.folder = folder;
    }

    public IDisposable Subscribe(IObserver<FileSystemEventArgs> observer)
    {
        // Inefficient if we get multiple subscribers.
        FileSystemWatcher watcher = new(this.folder);

        object sync = new();
        bool onErrorAlreadyCalled = false;

        void SendToObserver(object _, FileSystemEventArgs e)
        {
            lock (sync)
            {
                if (!onErrorAlreadyCalled)
                {
                    observer.OnNext(e); 
                }
            }
        }

        watcher.Created += SendToObserver;
        watcher.Changed += SendToObserver;
        watcher.Renamed += SendToObserver;
        watcher.Deleted += SendToObserver;

        watcher.Error += (_, e) =>
        {
            lock (sync)
            {
                // The FileSystemWatcher might report multiple errors, but
                // we're only allowed to report one to IObservable<T>.
                if (!onErrorAlreadyCalled)
                {

```

--------------------------------

### Create Observable from Task.Run

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Use Observable.FromAsync with Task.Run to create an observable that executes a task for each subscription. This ensures each subscriber gets a new task instance.

```csharp
IObservable<string> source = Observable.FromAsync(() => Task.Run(() =>
{
    Console.WriteLine("Task running...");
    return "Test";
}));
```

--------------------------------

### Manage disposable resources with Using

Source: https://context7.com/dotnet/reactive/llms.txt

Employ Using to tie the lifecycle of a disposable resource to an observable subscription, ensuring the resource is created lazily and disposed of automatically upon subscription completion or disposal.

```csharp
IObservable<string> lines = Observable.Using(
    resourceFactory:    () => File.OpenText("data.txt"),
    observableFactory:  reader =>
        Observable.Create<string>(async (observer, ct) =>
        {
            while (true)
            {
                string? line = await reader.ReadLineAsync(ct);
                if (line is null) { observer.OnCompleted(); return; }
                observer.OnNext(line);
            }
        }));
```

--------------------------------

### Creating a Simple Observable Sequence

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

This example demonstrates how to use Observable.Create to generate a sequence of integers. The provided delegate calls OnNext for each item and OnCompleted when the sequence finishes. It returns Disposable.Empty to indicate no specific unsubscription logic is needed.

```csharp
private IObservable<int> SomeNumbers()
{
    return Observable.Create<int>(
        (IObserver<int> observer) =>
        {
            observer.OnNext(1);
            observer.OnNext(2);
            observer.OnNext(3);
            observer.OnCompleted();

            return Disposable.Empty;
        });
}
```

--------------------------------

### Create Observable: Observable.Range

Source: https://context7.com/dotnet/reactive/llms.txt

Generates a cold observable sequence of integers within a specified range. The sequence starts from a given number and continues for a specified count.

```csharp
IObservable<int> range = Observable.Range(1, 5); // 1, 2, 3, 4, 5
range.Subscribe(Console.WriteLine, () => Console.WriteLine("Done"));
// Output: 1 2 3 4 5 Done
```

--------------------------------

### RefCount Behavior with Subscribers

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

This example demonstrates how RefCount manages connections and disconnections based on the number of active subscribers. The creation callback runs each time a new connection is established after a previous disconnection.

```csharp
var query = Observable.Create<int>(observer => {
    Console.WriteLine("Sub3: Connected");
    int count = 0;
    var timer = Observable.Interval(TimeSpan.FromSeconds(1)).Subscribe(x => {
        observer.OnNext(count++);
        Console.WriteLine($"Sub3: {count} ({DateTime.Now:HH:mm:ss})");
    });
    return () => {
        Console.WriteLine("Sub3: Disposed");
        timer.Dispose();
    };
}).Publish().RefCount();

var sub1 = query.Subscribe(x => Console.WriteLine($"Sub1: {x}"));
await Task.Delay(TimeSpan.FromSeconds(2));
var sub2 = query.Subscribe(x => Console.WriteLine($"Sub2: {x}"));
await Task.Delay(TimeSpan.FromSeconds(2));
sub1.Dispose();
await Task.Delay(TimeSpan.FromSeconds(2));
sub2.Dispose();
await Task.Delay(TimeSpan.FromSeconds(2));
var sub3 = query.Subscribe(x => Console.WriteLine($"Sub3: {x}"));
await Task.Delay(TimeSpan.FromSeconds(2));
sub3.Dispose();
```

--------------------------------

### Generate Range of Integers

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

Creates an observable sequence that produces a specified range of integers. The first integer is the start value, and the second is the count of values to yield.

```csharp
IObservable<int> range = Observable.Range(10, 15);
range.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
```

--------------------------------

### Demonstrating RefCount Subscription Management

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

Illustrates how RefCount manages subscriptions by automatically disposing of the underlying source when all subscribers are removed and reconnecting when new subscribers are added. This is shown by disposing of initial subscribers and then adding new ones.

```csharp
IDisposable s1 = rc.Subscribe(x => Console.WriteLine($"Sub1: {x} ({DateTime.Now})"));
IDisposable s2 = rc.Subscribe(x => Console.WriteLine($"Sub2: {x} ({DateTime.Now})"));
Thread.Sleep(600);

Console.WriteLine();
Console.WriteLine("Removing subscribers");
s1.Dispose();
s2.Dispose();
Thread.Sleep(600);

Console.WriteLine();
Console.WriteLine("Adding more subscribers");
Console.WriteLine();
rc.Subscribe(x => Console.WriteLine($"Sub3: {x} ({DateTime.Now})"));
rc.Subscribe(x => Console.WriteLine($"Sub4: {x} ({DateTime.Now})"));
```

--------------------------------

### Buffer Events into Chunks and Calculate Average Speed

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/08_Partitioning.md

This example uses Buffer to split navigation messages into chunks of 4 and then calculates the average speed across those chunks. If the source completes with fewer than the specified chunk size, the final chunk will be smaller.

```csharp
IObservable<IList<IVesselNavigation>> navigationChunks = 
   receiverHost.Messages.Where(v => v.Mmsi == 235009890)
                        .OfType<IVesselNavigation>()
                        .Where(n => n.SpeedOverGround.HasValue)
                        .Buffer(4);

IObservable<float> recentAverageSpeed = 
    navigationChunks.Select(chunk => chunk.Average(n => n.SpeedOverGround.Value));
```

--------------------------------

### Example Usage of Merge

Source: https://github.com/dotnet/reactive/blob/main/Rx.NET/Documentation/IntroToRx/09_CombiningSequences.md

Demonstrates how to use the `Merge` operator to combine two `IObservable<IAisMessage>` sequences from different AIS stations into a single stream of all messages.

```APIDOC
```csharp
IObservable<IAisMessage> station1 = aisStations.GetMessagesFromStation("AdurStation");
IObservable<IAisMessage> station2 = aisStations.GetMessagesFromStation("EastbourneStation");

IObservable<IAisMessage> allMessages = station1.Merge(station2);
```
```