Orleans 스트리밍 API

애플리케이션은 잘 알려진 .NET의 Rx(Reactive Extensions)와 매우 유사한 API를 통해 스트림과 상호 작용합니다. 주요 차이점은 Orleans 스트림 확장이 비동기적이므로 Orleans의 분산형 및 확장 가능한 컴퓨팅 패브릭에서 처리를 보다 효율적으로 수행할 수 있다는 것입니다.

비동기 스트림

애플리케이션은 스트림 공급자를 사용하여 스트림에 대한 핸들을 가져오는 것으로 시작합니다. 여기서 스트림 공급자에 대해 자세히 읽을 수 있지만 지금은 구현자가 스트림 동작 및 의미 체계를 사용자 지정할 수 있는 스트림 팩터리라고 생각할 수 있습니다.

IStreamProvider streamProvider = base.GetStreamProvider("SimpleStreamProvider");
StreamId streamId = StreamId.Create("MyStreamNamespace", Guid);
IAsyncStream<T> stream = streamProvider.GetStream<T>(streamId);

IStreamProvider streamProvider = base.GetStreamProvider("SimpleStreamProvider");
IAsyncStream<T> stream = streamProvider.GetStream<T>(Guid, "MyStreamNamespace");

애플리케이션은 조직 내부에 있을 때 Grain.GetStreamProvider 메서드를 호출하거나 클라이언트에서 GrainClient.GetStreamProvider 메서드를 호출하여 스트림 공급자에 대한 참조를 가져올 수 있습니다.

Orleans.Streams.IAsyncStream<T>은 가상 스트림에 대한 논리적이고 강력한 형식의 핸들입니다. Orleans 조직 참조와 개념이 유사합니다. GetStreamProvider 및 GetStream 호출은 순전히 로컬입니다. GetStream에 대한 인수는 GUID 및 스트림 네임스페이스(null일 수 있음)를 호출하는 추가 문자열입니다. GUID와 네임스페이스 문자열은 함께 스트림 ID를 구성합니다(IGrainFactory.GetGrain에 대한 인수와 개념이 유사함). GUID 및 네임스페이스 문자열의 조합은 스트림 ID를 결정할 때 추가적인 유연성을 제공합니다. 조직 7이 조직 형식 PlayerGrain 내에 존재할 수 있고 다른 조직 7이 조직 형식 ChatRoomGrain 내에 존재할 수 있는 것처럼 스트림 123은 스트림 네임스페이스 PlayerEventsStream와 함께 존재할 수 있으며 다른 스트림 123은 네임스페이스 ChatRoomMessagesStream 내에 존재할 수 있습니다.

생산 및 소비

IAsyncStream<T>은 IAsyncObserver<T> 및 IAsyncObservable<T> 인터페이스를 둘 다 구현합니다. 이렇게 하면 애플리케이션은 스트림을 통해 Orleans.Streams.IAsyncObserver<T>를 사용하여 스트림에 새 이벤트를 생성하거나 Orleans.Streams.IAsyncObservable<T>을 사용하여 스트림에서 이벤트를 구독하고 사용할 수 있습니다.

public interface IAsyncObserver<in T>
{
    Task OnNextAsync(T item, StreamSequenceToken token = null);
    Task OnCompletedAsync();
    Task OnErrorAsync(Exception ex);
}

public interface IAsyncObservable<T>
{
    Task<StreamSubscriptionHandle<T>> SubscribeAsync(IAsyncObserver<T> observer);
}

스트림에 이벤트를 생성하려면 애플리케이션을 호출하면 됩니다.

await stream.OnNextAsync<T>(event)

스트림을 구독하려면 애플리케이션을 호출합니다.

StreamSubscriptionHandle<T> subscriptionHandle = await stream.SubscribeAsync(IAsyncObserver)

SubscribeAsync에 대한 인수는 IAsyncObserver<T> 인터페이스를 구현하는 개체이거나 들어오는 이벤트를 처리하는 람다 함수의 조합일 수 있습니다. SubscribeAsync에 대한 추가 옵션은 AsyncObservableExtensions 클래스를 통해 사용할 수 있습니다. SubscribeAsync는 스트림에서 구독을 취소하는 데 사용할 수 있는 불투명 핸들인 StreamSubscriptionHandle<T>을 반환합니다(IDisposable의 비동기 버전과 개념이 유사함).

await subscriptionHandle.UnsubscribeAsync()

구독은 정품 인증이 아니라 조직용이라는 점에 유의해야 합니다. 조직 코드가 스트림에 구독되면 이 구독은 이 활성화의 수명을 초과하며, 조직 코드(잠재적으로 다른 활성화)가 명시적으로 구독을 취소할 때까지 영구적으로 지속됩니다. 이는 가상 스트림 추상화의 핵심입니다. 모든 스트림이 논리적으로 항상 존재할 뿐만 아니라 스트림 구독은 지속 가능하며 구독을 만든 특정 물리적 활성화 이후에도 지속됩니다.

다중성

Orleans 스트림에는 여러 생산자와 여러 소비자가 있을 수 있습니다. 생산자가 게시한 메시지는 메시지가 게시되기 전에 스트림을 구독한 모든 소비자에게 전달됩니다.

또한 소비자는 동일한 스트림을 여러 번 구독할 수 있습니다. 구독할 때마다 고유한 StreamSubscriptionHandle<T>을 반환합니다. 조직(또는 클라이언트)이 동일한 스트림에 대해 X번 구독하는 경우 각 구독에 대해 동일한 이벤트 X번을 한 번 수신합니다. 소비자는 개별 구독에서 구독을 취소할 수도 있습니다. 다음을 호출하여 현재 구독을 모두 찾을 수 있습니다.

IList<StreamSubscriptionHandle<T>> allMyHandles =
    await IAsyncStream<T>.GetAllSubscriptionHandles();

오류 복구

스트림의 생산자가 사라지거나 해당 조직이 비활성화된 경우 수행해야 할 작업은 없습니다. 다음에 이 조직이 더 많은 이벤트를 생성하려고 할 때 스트림 핸들을 다시 가져와 동일한 방식으로 새 이벤트를 생성할 수 있습니다.

소비자 논리는 조금 더 관련되어 있습니다. 앞서 설명한 것처럼 소비자 조직이 스트림을 구독하면 해당 조직이 명시적으로 구독을 취소할 때까지 유효합니다. 스트림의 소비자가 사라지거나(또는 해당 조직이 비활성화됨) 스트림에서 새 이벤트가 생성되면 소비자 조직이 자동으로 다시 활성화됩니다(메시지가 전송될 때 일반 Orleans 조직이 자동으로 활성화되는 것처럼). 이제 조직 코드에서 수행해야 하는 유일한 작업은 데이터를 처리할 수 있도록 IAsyncObserver<T>를 제공하는 것입니다. 소비자는 OnActivateAsync() 메서드의 일부로 처리 논리를 다시 연결해야 합니다. 이렇게 하려면 다음을 호출하면 됩니다.

StreamSubscriptionHandle<int> newHandle =
    await subscriptionHandle.ResumeAsync(IAsyncObserver);

소비자는 "처리 다시 시작"을 처음 구독할 때 가져온 이전 핸들을 사용합니다. ResumeAsync는 IAsyncObserver 논리의 새 인스턴스로 기존 구독을 업데이트할 뿐이며 이 소비자가 이미 이 스트림을 구독하고 있다는 사실을 변경하지 않습니다.

소비자는 어떻게 이전 subscriptionHandle을 가져오나요? 2가지 옵션이 있습니다. 소비자는 원래 SubscribeAsync 작업에서 다시 제공된 핸들을 유지했을 수 있으며 지금 사용할 수 있습니다. 또는 소비자에게 핸들이 없는 경우 다음을 호출하여 모든 활성 구독 핸들에 대해 IAsyncStream<T>을 요청할 수 있습니다.

IList<StreamSubscriptionHandle<T>> allMyHandles =
    await IAsyncStream<T>.GetAllSubscriptionHandles();

이제 소비자는 모든 것을 다시 시작하거나 원하는 경우 일부 구독을 취소할 수 있습니다.

팁

소비자 조직에서 IAsyncObserver<T> 인터페이스를 직접 구현하는 경우(public class MyGrain<T> : Grain, IAsyncObserver<T>) 이론적으로 IAsyncObserver를 다시 연결할 필요가 없으므로 ResumeAsync를 호출할 필요가 없습니다. 스트리밍 런타임은 해당 조직이 이미 IAsyncObserver를 구현하고 해당 IAsyncObserver 메서드를 호출한다는 것을 자동으로 파악할 수 있어야 합니다. 그러나 스트리밍 런타임은 현재 이를 지원하지 않으며, 조직이 직접 IAsyncObserver를 구현하더라도 조직 코드는 여전히 ResumeAsync를 명시적으로 호출해야 합니다.

명시적 및 암시적 구독

기본적으로 스트림 소비자는 스트림을 명시적으로 구독해야 합니다. 이 구독은 일반적으로 구독을 지시하는 조직(또는 클라이언트)이 수신하는 일부 외부 메시지에 의해 트리거됩니다. 예를 들어 채팅 서비스에서는 사용자가 채팅방에 조인할 때 조직은 채팅 이름으로 JoinChatGroup 메시지를 수신하므로 사용자 조직에서 이 채팅 스트림을 구독하게 됩니다.

또한 Orleans 스트림은 암시적 구독도 지원합니다. 이 모델에서 조직은 스트림을 명시적으로 구독하지 않습니다. 이 조직은 해당 조직 ID 및 ImplicitStreamSubscriptionAttribute에 따라 암시적으로 자동으로 구독됩니다. 암시적 구독의 기본 값은 스트림 활동이 조직 활성화를 트리거하도록 허용하여 구독을 자동으로 트리거하는 것입니다. 예를 들어 SMS 스트림을 사용하는 경우 한 조직이 스트림을 생성하고 다른 조직이 스트림을 처리하려면 생산자는 소비자 조직의 ID를 알고 스트림을 구독하라는 조직 호출을 수행해야 합니다. 그 후에만 이벤트 전송을 시작할 수 있습니다. 대신 암시적 구독을 사용하여 생산자는 스트림에 대한 이벤트 생성을 시작할 수 있으며 소비자 조직은 자동으로 활성화되고 스트림을 구독합니다. 이 경우 생산자는 누가 이벤트를 읽는지 신경 쓸 필요가 없습니다.

조직 구현 MyGrainType은 특성 [ImplicitStreamSubscription("MyStreamNamespace")]을 선언할 수 있습니다. 이는 ID가 GUID XXX 및 "MyStreamNamespace" 네임스페이스인 스트림에서 이벤트가 생성될 때 ID가 형식 MyGrainType의 XXX인 조직에 전달되어야 한다는 것을 스트리밍 런타임에 알려줍니다. 즉, 런타임은 스트림 <XXX, MyStreamNamespace>을 소비자 조직 <XXX, MyGrainType>에 매핑합니다.

ImplicitStreamSubscription이 존재하면 스트리밍 런타임이 이 조직을 스트림에 자동으로 구독하고 스트림 이벤트를 스트림에 전달합니다. 그러나 조직 코드는 이벤트를 처리하는 방법을 런타임에 알려야 합니다. 기본적으로 IAsyncObserver를 첨부해야 합니다. 따라서 조직이 활성화되면 OnActivateAsync 내부의 조직 코드는 다음을 호출해야 합니다.

IStreamProvider streamProvider =
    base.GetStreamProvider("SimpleStreamProvider");

StreamId streamId =
    StreamId.Create("MyStreamNamespace", this.GetPrimaryKey());
IAsyncStream<T> stream =
    streamProvider.GetStream<T>(streamId);

StreamSubscriptionHandle<T> subscription =
    await stream.SubscribeAsync(IAsyncObserver<T>);

IStreamProvider streamProvider =
    base.GetStreamProvider("SimpleStreamProvider");

IAsyncStream<T> stream =
    streamProvider.GetStream<T>(this.GetPrimaryKey(), "MyStreamNamespace");

StreamSubscriptionHandle<T> subscription =
    await stream.SubscribeAsync(IAsyncObserver<T>);

구독 논리 작성

다음은 명시적 및 암시적 구독, 되감기 및 되감기 불가능한 스트림과 같은 다양한 경우에 대한 구독 논리를 작성하는 방법에 대한 지침입니다. 명시적 구독과 암시적 구독의 주요 차이점은 암시적 구독의 경우 조직에는 모든 스트림 네임스페이스에 대해 항상 정확히 하나의 암시적 구독이 있다는 것입니다. 여러 구독을 만들 수 있는 방법이 없으며(구독 다중성이 없음), 구독을 취소할 방법이 없으며, 조직 논리는 항상 처리 논리만 연결하면 됩니다. 즉, 암시적 구독의 경우 구독을 다시 시작할 필요가 없습니다. 반면에 명시적 구독의 경우 구독을 다시 시작해야 합니다. 그렇지 않으면 조직이 다시 구독하면 조직이 여러 번 구독되는 결과가 초래됩니다.

암시적 구독:

암시적 구독의 경우 여전히 처리 논리를 연결하기 위해 구독해야 합니다. 이 작업은 소비자 조직에서 및 인터페이스를 구현하여 IStreamSubscriptionObserver 곡물이 구독과 IAsyncObserver<T> 별도로 활성화되도록 하여 수행할 수 있습니다. 스트림을 구독하기 위해 그레인은 핸들을 만들고 해당 OnSubscribed(...) 메서드에서 호출 await handle.ResumeAsync(this) 합니다.

메시지를 IAsyncObserver<T>.OnNextAsync(...) 처리하기 위해 스트림 데이터와 시퀀스 토큰을 수신하도록 메서드가 구현됩니다. 또는 메서드는 ResumeAsync 인터페이스onNextAsynconErrorAsync의 메서드 IAsyncObserver<T> 를 나타내는 대리자 집합을 사용할 수 있습니다onCompletedAsync.

public Task OnNextAsync(string item, StreamSequenceToken? token = null)
{
    _logger.LogInformation($"Received an item from the stream: {item}");
}

public async Task OnSubscribed(IStreamSubscriptionHandleFactory handleFactory)
{
    var handle = handleFactory.Create<string>();
    await handle.ResumeAsync(this);
}

public override async Task OnActivateAsync()
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var stream =
        streamProvider.GetStream<string>(
            this.GetPrimaryKey(), "MyStreamNamespace");

    await stream.SubscribeAsync(OnNextAsync);
}

명시적 구독:

명시적 구독의 경우 조직은 스트림을 구독하기 위해 SubscribeAsync를 호출해야 합니다. 그러면 구독이 생성되고 처리 논리가 연결됩니다. 명시적 구독은 조직이 구독을 취소할 때까지 존재하므로, 조직이 비활성화되었다가 다시 활성화되면 조직은 여전히 명시적으로 구독되지만 처리 논리는 연결되지 않습니다. 이 경우 조직은 처리 논리를 다시 연결해야 합니다. 이렇게 하려면 OnActivateAsync에서 조직은 먼저 IAsyncStream<T>.GetAllSubscriptionHandles()를 호출하여 어떤 구독을 가지고 있는지 알아내야 합니다. 조직은 처리를 계속하려는 각 핸들에서 ResumeAsync를 실행하거나 처리가 수행된 핸들에서 UnsubscribeAsync를 실행해야 합니다. 또한 조직은 필요에 따라 StreamSequenceToken을 ResumeAsync 호출에 대한 인수로 지정할 수 있습니다. 그러면 이 명시적 구독이 해당 토큰에서 소비를 시작합니다.

public async override Task OnActivateAsync(CancellationToken cancellationToken)
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var streamId = StreamId.Create("MyStreamNamespace", this.GetPrimaryKey());
    var stream = streamProvider.GetStream<string>(streamId);

    var subscriptionHandles = await stream.GetAllSubscriptionHandles();
    foreach (var handle in subscriptionHandles)
    {
       await handle.ResumeAsync(this);
    } 
}

public async override Task OnActivateAsync()
{
    var streamProvider = this.GetStreamProvider(PROVIDER_NAME);
    var stream =
        streamProvider.GetStream<string>(this.GetPrimaryKey(), "MyStreamNamespace");

    var subscriptionHandles = await stream.GetAllSubscriptionHandles();
    if (!subscriptionHandles.IsNullOrEmpty())
    {
        subscriptionHandles.ForEach(
            async x => await x.ResumeAsync(OnNextAsync));
    }
}

스트림 순서 및 시퀀스 토큰

개별 생산자와 개별 소비자 간의 이벤트 배달 순서는 스트림 공급자에 따라 달라집니다.

SMS를 통해 생산자는 생산자가 게시하는 방식을 제어하여 소비자가 볼 수 있는 이벤트의 순서를 명시적으로 제어합니다. 기본적으로(SMS 공급자에 대한 SimpleMessageStreamProviderOptions.FireAndForgetDelivery 옵션이 false로 설정된 경우) 생산자가 모든 OnNextAsync 호출을 기다리는 경우 이벤트는 FIFO 순서로 도착합니다. SMS에서 OnNextAsync 호출에 의해 반환된 손상된 Task로 표시되는 배달 오류를 처리하는 방법을 결정하는 것은 생산자에게 달려 있습니다.

기본 Azure 큐는 오류 발생 시 순서를 보장하지 않으므로 Azure 큐 스트림은 FIFO 순서를 보장하지 않습니다. (실패 없는 실행에서 FIFO 순서를 보장합니다.) 생산자가 Azure Queue로 이벤트를 생성할 때 큐 작업이 실패하는 경우 생산자가 다른 큐를 시도하고 나중에 잠재적 중복 메시지를 처리해야 합니다. 전달 측면에서 Orleans 스트리밍 런타임은 이벤트를 큐에서 제거하고 소비자에게 처리를 위해 전달하려고 시도합니다. Orleans 스트리밍 런타임은 성공적으로 처리될 때만 큐에서 이벤트를 삭제합니다. 전달 또는 처리에 실패하면 이벤트가 큐에서 삭제되지 않고 나중에 큐에 자동으로 다시 나타납니다. 스트리밍 런타임은 다시 전달하려고 시도하므로 FIFO 순서가 손상될 수 있습니다. 위의 동작은 Azure 큐의 일반적인 의미 체계와 일치합니다.

애플리케이션 정의 순서: 위의 순서 문제를 처리하기 위해 애플리케이션에서 필요에 따라 순서를 지정할 수 있습니다. 이는 이벤트를 정렬하는 데 사용할 수 있는 불투명 IComparable 개체인 StreamSequenceToken을 통해 수행됩니다. 생산자는 선택적 StreamSequenceToken을 OnNext 호출로 전달할 수 있습니다. 이 StreamSequenceToken은 소비자에게 전달되며 이벤트와 함께 전달됩니다. 이렇게 하면 애플리케이션이 스트리밍 런타임과 독립적으로 순서를 추론하고 재구성할 수 있습니다.

되감기 가능한 스트림

일부 스트림은 애플리케이션이 최신 시점부터 구독하도록 허용하는 반면, 다른 스트림은 "시간 되돌리기"를 허용합니다. 후자의 기능은 기본 큐 기술 및 특정 스트림 공급자에 따라 달라집니다. 예를 들어 Azure Queues는 큐에 넣은 최신 이벤트만 사용할 수 있지만 EventHub는 임의의 시점(일부 만료 시간까지)에서 이벤트를 재생할 수 있습니다. 시간 되돌리기를 지원하는 스트림을 되감기 가능한 스트림이라고 합니다.

되감기 가능한 스트림의 소비자는 StreamSequenceToken을 SubscribeAsync 호출로 전달할 수 있습니다. 런타임은 해당 StreamSequenceToken에서 시작하여 이벤트를 전달합니다. null 토큰은 소비자가 최신 이벤트부터 이벤트를 수신하기를 원한다는 의미입니다.

스트림을 되감는 기능은 복구 시나리오에서 매우 유용합니다. 예를 들어 스트림을 구독하고 주기적으로 최신 시퀀스 토큰과 함께 상태를 검사하는 조직을 고려합니다. 실패에서 복구할 때 조직은 최신 검사점 시퀀스 토큰에서 동일한 스트림을 다시 구독할 수 있으므로 마지막 검사점 이후 생성된 이벤트를 잃지 않고 복구할 수 있습니다.

Event Hubs 공급자는 되감기가 가능합니다. GitHub: Orleans/Azure/Orleans. Streaming.EventHubs에서 해당 코드를 찾을 수 있습니다. SMS 및 Azure Queue 공급자는 되감기할 수 없습니다.

상태 비저장 자동 스케일 아웃 처리

기본적으로 Orleans 스트리밍은 각각 하나 이상의 상태 저장 조직으로 처리되는 많은 수의 비교적 작은 스트림을 지원하는 것을 목표로 합니다. 전체적으로 모든 스트림의 처리는 많은 수의 일반(상태 저장) 조직 간에 분할됩니다. 애플리케이션 코드는 스트림 ID 및 조직 ID를 할당하고 명시적으로 구독하여 이 분할을 제어합니다. 목표는 분할된 상태 저장 처리입니다.

그러나 자동으로 스케일 아웃된 상태 비저장 처리의 흥미로운 시나리오도 있습니다. 이 시나리오에서 애플리케이션에는 적은 수의 스트림(또는 하나의 큰 스트림)이 있으며 목표는 상태 비저장 처리입니다. 예를 들어, 처리에는 각 이벤트를 디코딩하고 추가 상태 저장 처리를 위해 잠재적으로 다른 스트림으로 전달해야 하는 전역 이벤트 스트림이 있습니다. 상태 비저장 스케일 아웃 스트림 처리는 StatelessWorkerAttribute 조직을 통해 Orleans에서 지원될 수 있습니다.

상태 비저장 자동 스케일 아웃 처리의 현재 상태: 아직 구현되지 않았습니다. StatelessWorker 조직에서 스트림을 구독하려고 하면 정의되지 않은 동작이 발생합니다. 이 옵션을 지원하는 방안을 고려하고 있습니다.

조직 및 Orleans 클라이언트

Orleans 스트림은 조직 및 Orleans 클라이언트에서 균일하게 작동합니다. 즉, 동일한 API를 조직 내부와 Orleans 클라이언트에서 사용하여 이벤트를 생성하고 소비할 수 있습니다. 이렇게 하면 애플리케이션 논리가 크게 간소화되어 Grain Observers와 같은 특수 클라이언트 쪽 API가 중복됩니다.

완전 관리형 및 신뢰할 수 있는 스트리밍 pub-sub

스트림 구독을 추적하기 위해 Orleans(은)는 스트리밍 소비자 및 스트림 생산자를 위한 랑데부 지점 역할을 하는 Streaming Pub-Sub라는 런타임 구성 요소를 사용합니다. Pub-sub는 모든 스트림 구독을 추적하고 유지하며 스트림 소비자를 스트림 생산자와 일치시킵니다.

애플리케이션은 Pub-Sub 데이터가 저장되는 위치와 방법을 선택할 수 있습니다. Pub-Sub 구성 요소 자체는 Orleans 선언적 지속성을 사용하는 조직(PubSubRendezvousGrain이라고 함)으로 구현됩니다. PubSubRendezvousGrain은 PubSubStore라는 스토리지 공급자를 사용합니다. 다른 조직과 마찬가지로 스토리지 공급자에 대한 구현을 지정할 수 있습니다. 스트리밍 Pub-Sub의 경우 사일로 호스트 작성기를 사용하여 사일로 생성 시 PubSubStore 구현을 변경할 수 있습니다.

다음은 Azure 테이블에 상태를 저장하도록 Pub-Sub를 구성합니다.

hostBuilder.AddAzureTableGrainStorage("PubSubStore",
    options => options.ConfigureTableServiceClient("<Secret>"));

hostBuilder.AddAzureTableGrainStorage("PubSubStore",
    options => options.ConnectionString = "<Secret>");

이렇게 하면 Pub-Sub 데이터가 Azure Table에 영구적으로 저장됩니다. 초기 개발을 위해 메모리 스토리지도 사용할 수 있습니다. Pub-Sub 외에도 Orleans 스트리밍 런타임은 생산자에서 소비자에게 이벤트를 제공하고, 적극적으로 사용되는 스트림에 할당된 모든 런타임 리소스를 관리하며, 사용되지 않는 스트림에서 런타임 리소스를 투명하게 가비지 수집합니다.

구성

스트림을 사용하려면 사일로 호스트 또는 클러스터 클라이언트 빌더를 통해 스트림 공급자를 사용하도록 설정해야 합니다. 스트림 공급자에 대한 자세한 내용은 여기에서 확인할 수 있습니다. 샘플 스트림 공급자 설정:

hostBuilder.AddMemoryStreams("StreamProvider")
    .AddAzureQueueStreams<AzureQueueDataAdapterV2>("AzureQueueProvider",
        optionsBuilder => optionsBuilder.Configure(
            options => options.ConfigureTableServiceClient("<Secret>")))
    .AddAzureTableGrainStorage("PubSubStore",
        options => options.ConfigureTableServiceClient("<Secret>"));

hostBuilder.AddSimpleMessageStreamProvider("SMSProvider")
    .AddAzureQueueStreams<AzureQueueDataAdapterV2>("AzureQueueProvider",
        optionsBuilder => optionsBuilder.Configure(
            options => options.ConnectionString = "<Secret>"))
    .AddAzureTableGrainStorage("PubSubStore",
        options => options.ConnectionString = "<Secret>");

참고 항목

Orleans 스트림 공급자