-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add feedApi, feedConsumer templates (#88)
- Loading branch information
Showing
26 changed files
with
1,328 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| { | ||
| "$schema": "http://json.schemastore.org/template", | ||
| "author": "@jet @bartelink", | ||
| "classifications": [ | ||
| "Equinox", | ||
| "Propulsion", | ||
| "Event Sourcing", | ||
| "Feed" | ||
| ], | ||
| "tags": { | ||
| "language": "F#", | ||
| "type": "project" | ||
| }, | ||
| "identity": "FeedConsumer", | ||
| "name": "Propulsion FeedConsumer", | ||
| "shortName": "feedConsumer", | ||
| "sourceName": "FeedConsumerTemplate", | ||
| "preferNameDirectory": true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| module FeedConsumerTemplate.ApiClient | ||
|
|
||
| open FSharp.UMX | ||
| open System.Net.Http | ||
|
|
||
| open FeedConsumerTemplate.Domain | ||
|
|
||
| (* The feed presents a Tranche (series of epochs) per FC *) | ||
|
|
||
| module TrancheId = | ||
|
|
||
| let toFcId (x : Propulsion.Feed.TrancheId) : FcId = %x | ||
| let ofFcId (x : FcId) : Propulsion.Feed.TrancheId = %x | ||
|
|
||
| type TicketsEpochId = int<ticketsEpochId> | ||
| and [<Measure>] ticketsEpochId | ||
|
|
||
| [<NoComparison; NoEquality>] | ||
| type TicketsTranchesDto = { activeEpochs : TrancheReferenceDto[] } | ||
| and TrancheReferenceDto = { fc : FcId; epochId : TicketsEpochId } | ||
|
|
||
| (* Each Tranche response includes a checkpoint, which can be presented to Poll in order to resume consumption *) | ||
|
|
||
| type TicketsCheckpoint = int64<ticketsCheckpoint> | ||
| and [<Measure>] ticketsCheckpoint | ||
| module TicketsCheckpoint = | ||
| let ofPosition (x : Propulsion.Feed.Position) : TicketsCheckpoint = %x | ||
| let toPosition (x : TicketsCheckpoint) : Propulsion.Feed.Position = %x | ||
| let toStreamIndex (x : TicketsCheckpoint) : int64 = %x | ||
|
|
||
| type SliceDto = { closed : bool; tickets : TicketId[]; position : TicketsCheckpoint; checkpoint : TicketsCheckpoint } | ||
|
|
||
| type Session(client: HttpClient) = | ||
|
|
||
| member _.Send(req : HttpRequestMessage) : Async<HttpResponseMessage> = | ||
| client.Send(req) | ||
|
|
||
| type TicketsClient(session: Session) = | ||
|
|
||
| let basePath = "api/tickets" | ||
|
|
||
| member _.ActiveFcs() : Async<FcId[]> = async { | ||
| let request = HttpReq.get () |> HttpReq.withPath basePath | ||
| let! response = session.Send request | ||
| let! body = response |> HttpRes.deserializeOkJsonNet<TicketsTranchesDto> | ||
| return [| for f in body.activeEpochs -> f.fc |] | ||
| } | ||
|
|
||
| member _.ReadPage(fc : FcId, index : int) : Async<SliceDto> = async { | ||
| let request = HttpReq.post () |> HttpReq.withPathf "%s/%O/%d" basePath fc index | ||
| let! response = session.Send request | ||
| return! response |> HttpRes.deserializeOkJsonNet<SliceDto> | ||
| } | ||
|
|
||
| member _.Poll(fc : FcId, checkpoint: TicketsCheckpoint) : Async<SliceDto> = async { | ||
| let request = HttpReq.create () |> HttpReq.withPathf "%s/%O/slice/%O" basePath fc checkpoint | ||
| let! response = session.Send request | ||
| return! response |> HttpRes.deserializeOkJsonNet<SliceDto> | ||
| } | ||
|
|
||
| type Session with | ||
|
|
||
| member session.Tickets = TicketsClient session | ||
|
|
||
| type TicketsFeed(baseUri) = | ||
|
|
||
| let client = new HttpClient(BaseAddress = baseUri) | ||
| let tickets = Session(client).Tickets | ||
|
|
||
| // TODO add retries - consumer loop will abort if this throws | ||
| member _.Poll(trancheId, pos) : Async<Propulsion.Feed.Page<byte[]>> = async { | ||
| let checkpoint = TicketsCheckpoint.ofPosition pos | ||
| let! pg = tickets.Poll(TrancheId.toFcId trancheId, checkpoint) | ||
| let baseIndex = TicketsCheckpoint.toStreamIndex pg.position | ||
| let items = pg.tickets |> Array.mapi (fun i -> Ingester.PipelineEvent.ofIndexAndTicketId (baseIndex + int64 i)) | ||
| return { checkpoint = TicketsCheckpoint.toPosition pg.checkpoint; items = items; isTail = not pg.closed } | ||
| } | ||
|
|
||
| // TODO add retries - consumer loop will not commence if this emits an exception | ||
| member _.ReadTranches() : Async<Propulsion.Feed.TrancheId[]> = async { | ||
| let! activeFcs = tickets.ActiveFcs() | ||
| return [| for f in activeFcs -> TrancheId.ofFcId f |] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>netcoreapp3.1</TargetFramework> | ||
| <WarningLevel>5</WarningLevel> | ||
| <DisableImplicitFSharpCoreReference>true</DisableImplicitFSharpCoreReference> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <Compile Include="Infrastructure.fs" /> | ||
| <Compile Include="Types.fs" /> | ||
| <Compile Include="Ingester.fs" /> | ||
| <Compile Include="ApiClient.fs" /> | ||
| <Compile Include="Program.fs" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="FSharp.Core" Version="4.7.1" /> | ||
| <PackageReference Include="Argu" Version="6.1.1" /> | ||
| <PackageReference Include="Destructurama.FSharp" Version="1.1.1-dev-00033" /> | ||
| <PackageReference Include="Propulsion.Feed" Version="[2.10.0-rc10]" /> | ||
| <PackageReference Include="Equinox.CosmosStore" Version="3.0.0-beta.4" /> | ||
| <PackageReference Include="Propulsion.CosmosStore" Version="[2.10.0-rc10]" /> | ||
| <PackageReference Include="Serilog.Sinks.Console" Version="3.1.1" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
Oops, something went wrong.