egil

using Microsoft.Extensions.Logging;
using Orleans.Providers.Streams.Common;

// ReSharper disable once CheckNamespace
namespace Orleans.Providers.Streams;

internal sealed partial class DelayedStreamProviderStarter(
    string name,
    IControllable streamProvider,
    TimeSpan delayedStart,

egil

Using IStreamFilter in Orleans

This document provides comprehensive guidance on using IStreamFilter in Orleans based on research of the Orleans GitHub repository and official documentation.

Disclaimer: Created by Claude Sonnet 4.5 - AI makes mistakes, etc., not validated info. The prompt was as follows:

Investigate how to use IStreamFilter (need to check the dotnet/orleans codebase on github) to filter out specific events from certain stream providers. there is no documentation of the feature, so you need to go look at the source code in the orleans github repo and look at test cases that uses the feature to understand how it works. provide some sample code for me to look at

egil

public class ExampleTest(SiloFixture fixture) : IClassFixture<SiloFixture>
{
    [Fact]
    public async Task Sending_data_to_stream()
    {
        var message = "foo bar baz";
        var stream = fixture.GetStream<string>("grainId1");

        await stream.OnNextAsync(message);

egil

/// <summary>
/// Example implementation of an event-sourced grain that manages user state.
/// This grain demonstrates the key concepts of event sourcing:
/// 1. State is derived from a sequence of events rather than stored directly
/// 2. Each method that modifies state creates events that are processed
/// 3. The projection (User) is rebuilt by applying events through handlers
/// 4. Events can be processed individually or in batches with different transactional scopes
/// </summary>
public sealed class UserGrain(TimeProvider timeProvider) : EventGrain<UserGrain, User>(), IUserGrain
{

egil

public interface IExampleGrain : IStartupGrain, IGrainWithIntegerKey { }

public class ExampleGrain : Grain, IExampleGrain
{
    public override Task OnActivateAsync(CancellationToken cancellationToken)
    {
        // do things
    }
}

egil

public class ExampleTest(SiloFixture fixture) : IClassFixture<SiloFixture>
{
    [Fact]
    public async Task Sending_data_to_stream()
    {
        var message = "foo bar baz";
        var stream = fixture.GetStream<string>("grainId1");

        await stream.OnNextAsync(message);

egil

using Orleans.Storage;
using R3;

namespace Egil.Orleans.Storage.Testing;

public sealed class StorageObserver(string storageName, IGrainStorage observableTarget) : IGrainStorage
{
    private readonly Subject<StorageOperation> operationsFeed = new();

    public Observable<StorageOperation> StorageOperationFeed => operationsFeed.AsObservable();

egil

using System.Collections.Concurrent;
using System.Globalization;
using Orleans.Storage;
// https://www.nuget.org/packages/R3
using R3; 
using Xunit.Sdk;

namespace Egil.Orleans.Storage.Testing;

public sealed class ObservableMemoryStorage : IGrainStorage

egil

var builder = DistributedApplication.CreateBuilder(args);

var sqlPassword = builder.AddParameter("sql-password", secret: true);
var sqlserver = builder
    .AddSqlServer("sqlserver", password: sqlPassword)
    .WithDataVolume()
    .WithLifetime(ContainerLifetime.Persistent);

var importSourceConnectionString = "Server=[REPLACE].database.windows.net;Authentication=Active Directory Interactive;Database=[REPLACE];Encrypt=True;"
var db = sqlserver

egil

public sealed class RecordList<T> : List<T>, IEquatable<RecordList<T>>
    where T : IEquatable<T>
{
    private static readonly int EmptyRecordListHashCode = typeof(RecordList<T>).FullName!.GetHashCode(StringComparison.Ordinal);

    public RecordList()
    {
    }

    public RecordList(IEnumerable<T> collection)