Add .NET Workflow Streams samples

.github/CODEOWNERS

@@ -8,3 +8,7 @@
 # The Nexus team owns any folder whose name starts with "Nexus"
 /src/Nexus*/ @temporalio/nexus
 /tests/Nexus*/ @temporalio/nexus
+
+# SDK & AI SDK teams share ownership of Workflow Streams samples
+/src/WorkflowStreams/   @temporalio/sdk @temporalio/ai-sdk
+/tests/WorkflowStreams/ @temporalio/sdk @temporalio/ai-sdk

Directory.Packages.props

@@ -17,15 +17,17 @@
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.7.0" />
     <PackageVersion Include="Microsoft.VisualStudio.Threading.Analyzers" Version="18.7.23" />
     <PackageVersion Include="NexusRpc" Version="0.4.0" />
+    <PackageVersion Include="OpenAI" Version="2.13.0" />
     <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.16.0" />
     <PackageVersion Include="StyleCop.Analyzers" Version="1.2.0-beta.435" />
     <PackageVersion Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />
-    <PackageVersion Include="Temporalio" Version="1.18.0" />
+    <PackageVersion Include="Temporalio" Version="1.19.0" />
     <PackageVersion Include="Temporalio.Extensions.Aws.Lambda" Version="1.18.0" />
     <PackageVersion Include="Temporalio.Extensions.Aws.Lambda.OpenTelemetry" Version="1.18.0" />
     <PackageVersion Include="Temporalio.Extensions.DiagnosticSource" Version="1.18.0" />
     <PackageVersion Include="Temporalio.Extensions.Hosting" Version="1.18.0" />
     <PackageVersion Include="Temporalio.Extensions.OpenTelemetry" Version="1.18.0" />
+    <PackageVersion Include="Temporalio.Extensions.WorkflowStreams" Version="1.19.0" />
     <PackageVersion Include="TemporalCommunity.Aspire.Hosting" Version="0.1.0" />
     <PackageVersion Include="xunit" Version="2.9.3" />
     <PackageVersion Include="xunit.console" Version="2.9.3" />

README.md

@@ -53,6 +53,7 @@ Prerequisites:
 * [UpdateWithStartLazyInit](src/UpdateWithStartLazyInit) - Use update with start to lazily start a workflow before sending update.
 * [WorkerSpecificTaskQueues](src/WorkerSpecificTaskQueues) - Use a unique task queue per Worker to have certain Activities only run on that specific Worker.
 * [WorkerVersioning](src/WorkerVersioning) - How to use the Worker Versioning feature to more easily deploy changes to Workflow & other code.
+* [WorkflowStreams](src/WorkflowStreams) - Host a durable publish/subscribe stream inside a Workflow.
 * [WorkflowUpdate](src/WorkflowUpdate) - How to use the Workflow Update feature while blocking in update method for concurrent updates.
 
 ## Development

TemporalioSamples.sln

@@ -141,6 +141,18 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "SearchAttributes", "SearchA
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.SearchAttributes", "src\SearchAttributes\TemporalioSamples.SearchAttributes.csproj", "{97376F57-BA10-464B-AFBD-583E187DE947}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.BasicPublishSubscribe", "src\WorkflowStreams\BasicPublishSubscribe\TemporalioSamples.WorkflowStreams.BasicPublishSubscribe.csproj", "{091DC9D4-A88D-4393-9F84-32627DB1141F}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.BoundedLog", "src\WorkflowStreams\BoundedLog\TemporalioSamples.WorkflowStreams.BoundedLog.csproj", "{A1111111-1111-4111-8111-111111111111}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.ExternalPublisher", "src\WorkflowStreams\ExternalPublisher\TemporalioSamples.WorkflowStreams.ExternalPublisher.csproj", "{B2222222-2222-4222-8222-222222222222}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.ConcurrentSubscriptions", "src\WorkflowStreams\ConcurrentSubscriptions\TemporalioSamples.WorkflowStreams.ConcurrentSubscriptions.csproj", "{C3333333-3333-4333-8333-333333333333}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.LlmTokenStreaming", "src\WorkflowStreams\LlmTokenStreaming\TemporalioSamples.WorkflowStreams.LlmTokenStreaming.csproj", "{D4444444-4444-4444-8444-444444444444}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowStreams.ReconnectingSubscriber", "src\WorkflowStreams\ReconnectingSubscriber\TemporalioSamples.WorkflowStreams.ReconnectingSubscriber.csproj", "{E5555555-5555-4555-8555-555555555555}"
+EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "NexusStandaloneActivity", "NexusStandaloneActivity", "{5D5AFBCD-B4B7-4A4E-91C8-FD80E4C1E27B}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.NexusStandaloneActivity", "src\NexusStandaloneActivity\TemporalioSamples.NexusStandaloneActivity.csproj", "{4D8C9F9B-F8E3-4160-9286-32C3966A0125}"
@@ -803,6 +815,78 @@ Global
 		{97376F57-BA10-464B-AFBD-583E187DE947}.Release|x64.Build.0 = Release|Any CPU
 		{97376F57-BA10-464B-AFBD-583E187DE947}.Release|x86.ActiveCfg = Release|Any CPU
 		{97376F57-BA10-464B-AFBD-583E187DE947}.Release|x86.Build.0 = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|x64.Build.0 = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Debug|x86.Build.0 = Debug|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|x64.ActiveCfg = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|x64.Build.0 = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|x86.ActiveCfg = Release|Any CPU
+		{091DC9D4-A88D-4393-9F84-32627DB1141F}.Release|x86.Build.0 = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|x64.Build.0 = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Debug|x86.Build.0 = Debug|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|Any CPU.Build.0 = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|x64.ActiveCfg = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|x64.Build.0 = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|x86.ActiveCfg = Release|Any CPU
+		{A1111111-1111-4111-8111-111111111111}.Release|x86.Build.0 = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|x64.Build.0 = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Debug|x86.Build.0 = Debug|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|x64.ActiveCfg = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|x64.Build.0 = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|x86.ActiveCfg = Release|Any CPU
+		{B2222222-2222-4222-8222-222222222222}.Release|x86.Build.0 = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|x64.Build.0 = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Debug|x86.Build.0 = Debug|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|Any CPU.Build.0 = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|x64.ActiveCfg = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|x64.Build.0 = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|x86.ActiveCfg = Release|Any CPU
+		{C3333333-3333-4333-8333-333333333333}.Release|x86.Build.0 = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|x64.Build.0 = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Debug|x86.Build.0 = Debug|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|Any CPU.Build.0 = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|x64.ActiveCfg = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|x64.Build.0 = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|x86.ActiveCfg = Release|Any CPU
+		{D4444444-4444-4444-8444-444444444444}.Release|x86.Build.0 = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|x64.Build.0 = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Debug|x86.Build.0 = Debug|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|x64.ActiveCfg = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|x64.Build.0 = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|x86.ActiveCfg = Release|Any CPU
+		{E5555555-5555-4555-8555-555555555555}.Release|x86.Build.0 = Release|Any CPU
 		{4D8C9F9B-F8E3-4160-9286-32C3966A0125}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{4D8C9F9B-F8E3-4160-9286-32C3966A0125}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{4D8C9F9B-F8E3-4160-9286-32C3966A0125}.Debug|x64.ActiveCfg = Debug|Any CPU
@@ -885,6 +969,12 @@ Global
 		{6B6622AE-2970-4AAD-B2E4-A7EE1E2C40EA} = {17436B0C-8853-030B-9E2F-BAEB65BE59FB}
 		{34ADAC00-5559-6993-F779-D71CE5A0F93F} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 		{97376F57-BA10-464B-AFBD-583E187DE947} = {34ADAC00-5559-6993-F779-D71CE5A0F93F}
+		{091DC9D4-A88D-4393-9F84-32627DB1141F} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{A1111111-1111-4111-8111-111111111111} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{B2222222-2222-4222-8222-222222222222} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{C3333333-3333-4333-8333-333333333333} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{D4444444-4444-4444-8444-444444444444} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{E5555555-5555-4555-8555-555555555555} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 		{5D5AFBCD-B4B7-4A4E-91C8-FD80E4C1E27B} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 		{4D8C9F9B-F8E3-4160-9286-32C3966A0125} = {5D5AFBCD-B4B7-4A4E-91C8-FD80E4C1E27B}
 	EndGlobalSection

src/WorkflowStreams/BasicPublishSubscribe/Constants.cs

@@ -0,0 +1,12 @@
+namespace TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+public static class Constants
+{
+    public const string TaskQueue = "workflow-streams-basic-publish-subscribe";
+
+    public const string TopicStatus = "status";
+
+    public const string TopicProgress = "progress";
+
+    public static readonly TimeSpan DrainDelay = TimeSpan.FromMilliseconds(500);
+}

src/WorkflowStreams/BasicPublishSubscribe/Models.cs

@@ -0,0 +1,9 @@
+namespace TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+using Temporalio.Extensions.WorkflowStreams;
+
+public record OrderInput(string OrderId, WorkflowStreamState? StreamState = null);
+
+public record StatusEvent(string Kind, string OrderId);
+
+public record ProgressEvent(string Message);

src/WorkflowStreams/BasicPublishSubscribe/OrderWorkflow.workflow.cs

@@ -0,0 +1,31 @@
+namespace TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+using Temporalio.Extensions.WorkflowStreams;
+using Temporalio.Workflows;
+
+[Workflow]
+public class OrderWorkflow
+{
+    private readonly WorkflowStream stream;
+
+    [WorkflowInit]
+    public OrderWorkflow(OrderInput input) => stream = new(input.StreamState);
+
+    [WorkflowRun]
+    public async Task<string> RunAsync(OrderInput input)
+    {
+        var status = stream.GetTopic<StatusEvent>(Constants.TopicStatus);
+        var progress = stream.GetTopic<ProgressEvent>(Constants.TopicProgress);
+
+        status.Publish(new StatusEvent("received", input.OrderId));
+        var chargeId = await Workflow.ExecuteActivityAsync(
+            () => PaymentActivities.ChargeCardAsync(input.OrderId),
+            new() { StartToCloseTimeout = TimeSpan.FromMinutes(1), });
+        status.Publish(new StatusEvent("shipped", input.OrderId));
+        progress.Publish(new ProgressEvent($"charge id: {chargeId}"));
+        status.Publish(new StatusEvent("complete", input.OrderId));
+
+        await Workflow.DelayAsync(Constants.DrainDelay);
+        return chargeId;
+    }
+}

src/WorkflowStreams/BasicPublishSubscribe/PaymentActivities.cs

@@ -0,0 +1,26 @@
+namespace TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+using Microsoft.Extensions.Logging;
+using Temporalio.Activities;
+using Temporalio.Extensions.WorkflowStreams;
+
+public static class PaymentActivities
+{
+    [Activity]
+    public static async Task<string> ChargeCardAsync(string orderId)
+    {
+        await using var streamClient = WorkflowStreamClient.FromActivity(
+            new() { BatchInterval = TimeSpan.FromMilliseconds(200), });
+        var progress = streamClient.GetTopic<ProgressEvent>(Constants.TopicProgress);
+
+        progress.Publish(new ProgressEvent("charging card..."));
+        ActivityExecutionContext.Current.Logger.LogInformation(
+            "Charging card for order {OrderId}", orderId);
+        await Task.Delay(
+            TimeSpan.FromSeconds(1),
+            ActivityExecutionContext.Current.CancellationToken);
+        progress.Publish(new ProgressEvent("card charged"));
+        await streamClient.FlushAsync();
+        return $"charge-{orderId}";
+    }
+}

src/WorkflowStreams/BasicPublishSubscribe/Program.cs

@@ -0,0 +1,33 @@
+using Microsoft.Extensions.Logging;
+using Temporalio.Client;
+using Temporalio.Common.EnvConfig;
+using Temporalio.Worker;
+using TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+var connectOptions = ClientEnvConfig.LoadClientConnectOptions();
+connectOptions.TargetHost ??= "localhost:7233";
+connectOptions.LoggerFactory = LoggerFactory.Create(builder =>
+    builder.
+        AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").
+        SetMinimumLevel(LogLevel.Information));
+var client = await TemporalClient.ConnectAsync(connectOptions);
+
+if (args.ElementAtOrDefault(0) == "worker")
+{
+    using var tokenSource = new CancellationTokenSource();
+    Console.CancelKeyPress += (_, eventArgs) =>
+    {
+        tokenSource.Cancel();
+        eventArgs.Cancel = true;
+    };
+    using var worker = new TemporalWorker(
+        client,
+        new TemporalWorkerOptions(Constants.TaskQueue).
+            AddActivity(PaymentActivities.ChargeCardAsync).
+            AddWorkflow<OrderWorkflow>());
+    await worker.ExecuteAsync(tokenSource.Token);
+}
+else
+{
+    await Scenario.RunPublisherAsync(client);
+}

src/WorkflowStreams/BasicPublishSubscribe/README.md

@@ -0,0 +1,16 @@
+# Basic publish/subscribe
+
+An order Workflow publishes lifecycle events to `status`, while its payment Activity publishes
+progress events to `progress`. The subscriber consumes and decodes both topics.
+
+Start a Temporal service, then run the worker:
+
+```bash
+dotnet run -- worker
+```
+
+In another terminal, run the subscriber:
+
+```bash
+dotnet run
+```

src/WorkflowStreams/BasicPublishSubscribe/Scenario.cs

@@ -0,0 +1,49 @@
+namespace TemporalioSamples.WorkflowStreams.BasicPublishSubscribe;
+
+using Temporalio.Client;
+using Temporalio.Converters;
+using Temporalio.Extensions.WorkflowStreams;
+
+public static class Scenario
+{
+    public static async Task RunPublisherAsync(ITemporalClient client)
+    {
+        var workflowId = $"workflow-streams-order-{Guid.NewGuid()}";
+        var handle = await client.StartWorkflowAsync(
+            (OrderWorkflow wf) => wf.RunAsync(new OrderInput("order-42", null)),
+            new(workflowId, Constants.TaskQueue));
+        Console.WriteLine($"Started workflow: {workflowId}");
+
+        await using var streamClient = new WorkflowStreamClient(client, workflowId);
+        var options = new WorkflowStreamSubscribeOptions
+        {
+            Topics = new List<string>
+            {
+                Constants.TopicStatus,
+                Constants.TopicProgress,
+            },
+        };
+        await foreach (var item in streamClient.SubscribeAsync(options))
+        {
+            if (item.Topic == Constants.TopicStatus)
+            {
+                var evt = Decode<StatusEvent>(client, item);
+                Console.WriteLine($"[status]   {evt.Kind}: order={evt.OrderId}");
+                if (evt.Kind == "complete")
+                {
+                    break;
+                }
+            }
+            else if (item.Topic == Constants.TopicProgress)
+            {
+                var evt = Decode<ProgressEvent>(client, item);
+                Console.WriteLine($"[progress] {evt.Message}");
+            }
+        }
+
+        Console.WriteLine($"Workflow result: {await handle.GetResultAsync()}");
+    }
+
+    private static T Decode<T>(ITemporalClient client, WorkflowStreamItem item) =>
+        client.Options.DataConverter.PayloadConverter.ToValue<T>(item.Payload);
+}

src/WorkflowStreams/BasicPublishSubscribe/TemporalioSamples.WorkflowStreams.BasicPublishSubscribe.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Temporalio.Extensions.WorkflowStreams" />
+  </ItemGroup>
+
+</Project>

src/WorkflowStreams/BoundedLog/Constants.cs

@@ -0,0 +1,10 @@
+namespace TemporalioSamples.WorkflowStreams.BoundedLog;
+
+public static class Constants
+{
+    public const string TaskQueue = "workflow-streams-bounded-log";
+
+    public const string TopicTick = "tick";
+
+    public static readonly TimeSpan DrainDelay = TimeSpan.FromMilliseconds(500);
+}

src/WorkflowStreams/BoundedLog/Models.cs

@@ -0,0 +1,12 @@
+namespace TemporalioSamples.WorkflowStreams.BoundedLog;
+
+using Temporalio.Extensions.WorkflowStreams;
+
+public record TickerInput(
+    int Count = 50,
+    int KeepLast = 10,
+    int TruncateEvery = 5,
+    TimeSpan? Interval = null,
+    WorkflowStreamState? StreamState = null);
+
+public record TickEvent(int N);

src/WorkflowStreams/BoundedLog/Program.cs

@@ -0,0 +1,31 @@
+using Microsoft.Extensions.Logging;
+using Temporalio.Client;
+using Temporalio.Common.EnvConfig;
+using Temporalio.Worker;
+using TemporalioSamples.WorkflowStreams.BoundedLog;
+
+var connectOptions = ClientEnvConfig.LoadClientConnectOptions();
+connectOptions.TargetHost ??= "localhost:7233";
+connectOptions.LoggerFactory = LoggerFactory.Create(builder =>
+    builder.
+        AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").
+        SetMinimumLevel(LogLevel.Information));
+var client = await TemporalClient.ConnectAsync(connectOptions);
+
+if (args.ElementAtOrDefault(0) == "worker")
+{
+    using var tokenSource = new CancellationTokenSource();
+    Console.CancelKeyPress += (_, eventArgs) =>
+    {
+        tokenSource.Cancel();
+        eventArgs.Cancel = true;
+    };
+    using var worker = new TemporalWorker(
+        client,
+        new TemporalWorkerOptions(Constants.TaskQueue).AddWorkflow<TickerWorkflow>());
+    await worker.ExecuteAsync(tokenSource.Token);
+}
+else
+{
+    await Scenario.RunTruncatingTickerAsync(client);
+}