Skip to content

Commit 43f1bd1

Browse files
authored
Update to Propulsion 2.12.0-rc.2 (#107)
1 parent f5016b6 commit 43f1bd1

File tree

22 files changed

+110
-97
lines changed

22 files changed

+110
-97
lines changed

CHANGELOG.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@ The `Unreleased` section name is replaced by the expected version of next releas
1818
### Changed
1919

2020
- Remove usage of `type Command` DUs [#103](https://github.com/jet/dotnet-templates/pull/103)
21-
- Target `Equinox` v `3.0.5`
21+
- Target `Equinox` v `3.0.5`, `Propulsion` v `2.12.0-rc.2`
22+
- Target `Destructurama.FSharp` v `1.2.0`, `Serilog.Sinks.Async` v `1.5.0`, `Serilog.Sinks.Console` v `4.0.0` [#101](https://github.com/jet/dotnet-templates/pull/101)
2223

2324
### Removed
2425
### Fixed

equinox-shipping/Watchdog.Integration/ReactorFixture.fs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ type MemoryReactorFixture(testOutput) =
1313
let processManager = Shipping.Domain.FinalizationWorkflow.Config.create 4 store.Config
1414
let sink =
1515
let processingTimeout = TimeSpan.FromSeconds 1.
16-
let maxReadAhead, maxConcurrentStreams = 1024, 4 // TODO make Int32.MaxValue work
16+
let maxReadAhead, maxConcurrentStreams = Int32.MaxValue, 4
1717
Program.startWatchdog log (processingTimeout, stats) (maxReadAhead, maxConcurrentStreams) processManager.Pump
1818
let projector = MemoryStoreProjector.Start(log, sink)
1919
let projectorStoreSubscription =

equinox-shipping/Watchdog/Program.fs

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -123,17 +123,18 @@ let build (args : Args.Arguments) =
123123
let processManager = Shipping.Domain.FinalizationWorkflow.Config.create args.ProcessManagerMaxDop storeCfg
124124
let stats = Handler.Stats(Log.Logger, statsInterval=args.StatsInterval, stateInterval=args.StateInterval)
125125
startWatchdog Log.Logger (args.ProcessingTimeout, stats) (maxReadAhead, maxConcurrentStreams) processManager.Pump
126-
let pipeline =
127-
use observer = CosmosStoreSource.CreateObserver(Log.Logger, sink.StartIngester, Seq.collect Handler.transformOrFilter)
126+
let source =
127+
let observer = CosmosStoreSource.CreateObserver(Log.Logger, sink.StartIngester, Seq.collect Handler.transformOrFilter)
128128
let leases, startFromTail, maxItems, lagFrequency = args.Cosmos.MonitoringParams()
129-
CosmosStoreSource.Run(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxItems=maxItems, lagReportFreq=lagFrequency)
130-
sink, pipeline
131-
132-
let run args = async {
133-
let sink, pipeline = build args
134-
pipeline |> Async.Start
135-
return! sink.AwaitWithStopOnCancellation()
136-
}
129+
CosmosStoreSource.Start(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxItems=maxItems, lagReportFreq=lagFrequency)
130+
sink, source
131+
132+
open Propulsion.CosmosStore.Infrastructure // AwaitKeyboardInterruptAsTaskCancelledException
133+
134+
let run args =
135+
let sink, source = build args
136+
[ Async.AwaitKeyboardInterruptAsTaskCancelledException(); source.AwaitWithStopOnCancellation(); sink.AwaitWithStopOnCancellation() ]
137+
|> Async.Parallel |> Async.Ignore<unit[]>
137138

138139
[<EntryPoint>]
139140
let main argv =

equinox-shipping/Watchdog/Watchdog.fsproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
<ItemGroup>
1717
<PackageReference Include="Argu" Version="6.1.1" />
1818
<PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
19-
<PackageReference Include="Propulsion.CosmosStore" Version="2.11.0" />
19+
<PackageReference Include="Propulsion.CosmosStore" Version="2.12.0-rc.2" />
2020
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
2121
</ItemGroup>
2222

feed-consumer/FeedConsumer.fsproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919
<PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
2020
<PackageReference Include="Equinox.CosmosStore" Version="3.0.5" />
2121
<PackageReference Include="Equinox.CosmosStore.Prometheus" Version="3.0.5" />
22-
<PackageReference Include="Propulsion.CosmosStore" Version="2.11.0" />
23-
<PackageReference Include="Propulsion.Feed" Version="2.11.0" />
22+
<PackageReference Include="Propulsion.CosmosStore" Version="2.12.0-rc.2" />
23+
<PackageReference Include="Propulsion.Feed" Version="2.12.0-rc.2" />
2424
<PackageReference Include="Serilog.Sinks.Async" Version="1.5.0" />
2525
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
2626
</ItemGroup>

periodic-ingester/ApiClient.fs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,5 +32,5 @@ type TicketsFeed(baseUri) =
3232
let tickets = TicketsClient(client)
3333

3434
// TODO add retries - consumer loop will abort if this throws
35-
member _.Crawl(): AsyncSeq<Propulsion.Feed.SourceItem[]> =
35+
member _.Crawl(_trancheId): AsyncSeq<Propulsion.Feed.SourceItem[]> =
3636
tickets.Crawl()

periodic-ingester/PeriodicIngester.fsproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
<PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
2121
<PackageReference Include="Equinox.CosmosStore.Prometheus" Version="3.0.5" />
2222
<PackageReference Include="prometheus-net.AspNetCore" Version="3.6.0" />
23-
<PackageReference Include="Propulsion.CosmosStore" Version="2.11.0" />
24-
<PackageReference Include="Propulsion.Feed" Version="2.11.0" />
23+
<PackageReference Include="Propulsion.CosmosStore" Version="2.12.0-rc.2" />
24+
<PackageReference Include="Propulsion.Feed" Version="2.12.0-rc.2" />
2525
<PackageReference Include="Serilog.Sinks.Async" Version="1.5.0" />
2626
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
2727
</ItemGroup>

propulsion-archiver/Archiver.fsproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
<PackageReference Include="Argu" Version="6.1.1" />
1717
<PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
1818
<PackageReference Include="Equinox.CosmosStore.Prometheus" Version="3.0.5" />
19-
<PackageReference Include="Propulsion.CosmosStore" Version="2.11.0" />
19+
<PackageReference Include="Propulsion.CosmosStore" Version="2.12.0-rc.2" />
2020
<PackageReference Include="prometheus-net.AspNetCore" Version="3.6.0" />
2121
<PackageReference Include="Serilog.Sinks.Async" Version="1.5.0" />
2222
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />

propulsion-archiver/Program.fs

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -184,11 +184,11 @@ let build (args : Args.Arguments, log) =
184184
let context = args.DestinationArchive.Connect() |> Async.RunSynchronously |> CosmosStoreContext.create
185185
let eventsContext = Equinox.CosmosStore.Core.EventsContext(context, Config.log)
186186
CosmosStoreSink.Start(log, args.MaxReadAhead, eventsContext, args.MaxWriters, args.StatsInterval, args.StateInterval, (*purgeInterval=TimeSpan.FromMinutes 10.,*) maxBytes = args.MaxBytes)
187-
let pipeline =
188-
use observer = CosmosStoreSource.CreateObserver(log, archiverSink.StartIngester, Seq.collect Handler.selectArchivable)
187+
let source =
188+
let observer = CosmosStoreSource.CreateObserver(log, archiverSink.StartIngester, Seq.collect Handler.selectArchivable)
189189
let monitored, leases, processorName, startFromTail, maxItems, lagFrequency = args.MonitoringParams()
190-
CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxItems=maxItems, lagReportFreq=lagFrequency)
191-
archiverSink, pipeline
190+
CosmosStoreSource.Start(log, monitored, leases, processorName, observer, startFromTail, ?maxItems=maxItems, lagReportFreq=lagFrequency)
191+
archiverSink, source
192192

193193
// A typical app will likely have health checks etc, implying the wireup would be via `endpoints.MapMetrics()` and thus not use this ugly code directly
194194
let startMetricsServer port : IDisposable =
@@ -197,12 +197,13 @@ let startMetricsServer port : IDisposable =
197197
Log.Information("Prometheus /metrics endpoint on port {port}", port)
198198
{ new IDisposable with member x.Dispose() = ms.Stop(); (metricsServer :> IDisposable).Dispose() }
199199

200+
open Propulsion.CosmosStore.Infrastructure // AwaitKeyboardInterruptAsTaskCancelledException
201+
200202
let run args = async {
201203
let log = Log.ForContext<Propulsion.Streams.Scheduling.StreamSchedulingEngine>()
202-
let sink, pipeline = build (args, log)
203-
pipeline |> Async.Start
204+
let sink, source = build (args, log)
204205
use _metricsServer : IDisposable = args.PrometheusPort |> Option.map startMetricsServer |> Option.toObj
205-
do! sink.AwaitWithStopOnCancellation()
206+
return! Async.Parallel [ Async.AwaitKeyboardInterruptAsTaskCancelledException(); source.AwaitWithStopOnCancellation(); sink.AwaitWithStopOnCancellation() ] |> Async.Ignore<unit[]>
206207
}
207208

208209
[<EntryPoint>]

propulsion-consumer/Consumer.fsproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
<ItemGroup>
1717
<PackageReference Include="Argu" Version="6.1.1" />
1818
<PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
19-
<PackageReference Include="Propulsion.Kafka" Version="2.11.0" />
19+
<PackageReference Include="Propulsion.Kafka" Version="2.12.0-rc.2" />
2020
<PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
2121
</ItemGroup>
2222

0 commit comments

Comments
 (0)