Modeling Pub/Sub system with F# and CQRS (on micro-scale)

Almost all APIs that are built in different companies have a lot in common, but when you have to open several projects, they are very different. Query and error handling, interaction with domain objects and infrastructure layers, etc. shouldn't be so different. All this fake difference only complicates the understanding of the components of the project and their interaction. In this talk, I will share how we build APIs that support different semantic models(Request/Response and Pub/Sub) and various protocols exchange (HTTP/WebSockets/AMQP). I will also try to unlock the potential of the CQRS architectural pattern and explain why we apply it to almost all API projects.

1. Modeling Pub/Sub system with F# and CQRS (on micro scale) by @antyadev
2. The Problem Almost all APIs that are built in different companies have a lot in common, but when you have to open several projects, they are very different. Query and error handling, interaction with domain objects and infrastructure layers, etc. shouldn't be so different. All this fake difference only complicates the understanding of the components of the project and their interaction.
3. @antyadev antyadev@gmail.com
4. Data Feed WebSites/Terminals
5. Data Feed F# F# WebSites/Terminals F#
6. Agenda CQRS in 5 mins F# and CQRS(on micro scale) Advantages and disadvantages
7. CQRS
13. CQRS (on micro scale)
14. Incidents
15. WebScokets DB
16. Subscribe
18. { incident }
19. push { incident }
20. Command Command
21. Command Event
22. Incidents Domain
23. type IncidentType = Goal | Corner | Penalty | RedCard | YellowCard type ParticipantRole = Home | Away type Incident = { Id: IncidentId GameId: GameId Type: IncidentType ParticipantId: ParticipantId ParticipantRole: ParticipantRole option TimeInGame: DateTime GamePart: GamePart option }
24. type Query = | Incident of IncidentId | IncidentsForGames of GameId[] * IncidentType[] * LimitsParams type Command = | AddIncident of IncidentArgs | RemoveIncident of IncidentId | Subscribe of GameId[] * IncidentType[] | Unsubscribe of SubscriptionId type DomainEvent = | IncidentAdded of Incident * subscriptionsToNotify:SubscriptionId[] | IncidentRemoved of IncidentId | SubscriptionCreated of SubscriptionId | SubscriberAdded of SubscriptionId * subscribersCount:uint32 | SubscriptionRemoved of SubscriptionId | SubscriberRemoved of SubscriptionId * subscribersCount:uint32
25. type ParserError = | GameIdFormat of originValues:string[] | IncidentIdFormat of originValues:string[] | IncidentTypeFormat of originValues:string[] type DomainError = | IncidentCantBeCreated of gameId:int * typeId:int | LimitsOutOfRange of top:int * maxTop:int * skip:int | GameNotFound of gameId:int | IncidentNotFound of incidentId:int | IncidentsForGamesNotFound of gameIds:int[] type AppError = | Parser of ParserError | Domain of DomainError static member create(e: ParserError) = e |> Parser static member create(e: DomainError) = e |> Domain static member createResult(e: DomainError) = e |> Domain |> Error static member createResult(e: ParserError) = e |> Parser |> Error
26. Project development with familiar backbone Domain | Commands | Queries | Events CommandHandler | QueryHandler CQRS (on micro scale)
27. In every project you will see the following structure CQRS (on micro scale)
28. Domain Queries Commands Events Errors Command Handler Query Handler
29. Command Handler CQRS (on micro scale)
30. Command Handler Command -> DomainEvent
31. Command Handler Command -> Result<DomainEvent, AppError>
32. Command Handler Command -> Async<Result<DomainEvent, AppError>>
33. Command Handler Command -> Async<Result<DomainEvent, AppError>> TOP LEVEL CONSTRAINT which force any new functionality to return result
34. type Command = | AddIncident of IncidentArgs | RemoveIncident of IncidentId | Subscribe of GameId[] * IncidentType[] | Unsubscribe of SubscriptionId type DomainEvent = | IncidentAdded of Incident * subscriptionsToNotify:SubscriptionId[] | IncidentRemoved of IncidentId | SubscriptionCreated of SubscriptionId | SubscriberAdded of SubscriptionId * subscribersCount:uint32 | SubscriptionRemoved of SubscriptionId | SubscriberRemoved of SubscriptionId * subscribersCount:uint32
35. let handle (command) = match command with | AddIncident args -> DomainEvent.IncidentAdded | RemoveIncident id -> DomainEvent.IncidentRemoved | Subscribe (gamesIds, incTypes) -> DomainEvent.SubscriptionCreated | Unsubscribe id -> DomainEvent.SubscriptionRemoved type Command = | AddIncident of IncidentArgs | RemoveIncident of IncidentId | Subscribe of GameId[] * IncidentType[] | Unsubscribe of SubscriptionId type DomainEvent = | IncidentAdded of Incident * subscriptionsToNotify:SubscriptionId[] | IncidentRemoved of IncidentId | SubscriptionCreated of SubscriptionId | SubscriberAdded of SubscriptionId * subscribersCount:uint32 | SubscriptionRemoved of SubscriptionId | SubscriberRemoved of SubscriptionId * subscribersCount:uint32
36. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = let handle (command) = asyncResult { match command with | AddIncident params -> let! game = storage.FindGame(params.GametId) with match game with | Some game -> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle(command) = handle(command)
37. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = let handle (command) = asyncResult { match command with | AddIncident params -> let! game = storage.FindGame(params.GametId) with match game with | Some game -> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle
38. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = let handle (command) = asyncResult { match command with | AddIncident params -> match! storage.FindGame(params.GametId) with | Some game -> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle
39. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = // handle: Command -> Async<Result<DomainEvent, AppError>> let handle (command) = asyncResult { match command with | AddIncident params -> match! storage.FindGame(params.GametId) with // Task<Option<Game>> | Some game -> // Result<DomainEvent, AppError> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle
40. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = // handle: Command -> Async<Result<DomainEvent, AppError>> let handle (command) = asyncResult { match command with | CreateIncident params -> match! storage.FindGame(params.GametId) with // Task<Option<Game>> | Some game -> // Result<DomainEvent, AppError> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle
41. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = let eventStream = new Subject<DomainEvent>() let publishEvent (event) = eventStream.OnNext(event) // handle: Command -> Async<Result<DomainEvent, AppError>> let handle (command) = asyncResult { match command with | CreateIncident params -> match! storage.FindGame(params.GametId) with // Task<Option<Game>> | Some game -> // Result<DomainEvent, AppError> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle member x.EventStream = eventStream :> IObservable<DomainEvent>
42. type CommandHandler(storage: IStorage, subscriptions: Subscriptions) = let eventStream = new Subject<DomainEvent>() let publishEvent = eventStream.OnNext // handle: Command -> Async<Result<DomainEvent, AppError>> let handle (command) = asyncResult { match command with | CreateIncident params -> match! storage.FindGame(params.GametId) with // Task<Option<Game>> | Some game -> // Result<DomainEvent, AppError> let! event = IncidentDomain.add(params, game, subscriptions) publishEvent(event) return event | None -> return! GameNotFound(params.GametId) |> AppError.create } member x.Handle = handle member x.EventStream = eventStream :> IObservable<DomainEvent>
43. Command Handler Handle: Command -> Async<Result<DomainEvent, AppError> EventStream: IObservable<DomainEvent> WebScokets
44. Command Handler WebSockets Middleware AMQP Middleware connections connections EventStream: IObservable<DomainEvent> Dictionary<SubscriptionId, Connections[]> Dictionary<SubscriptionId, Connections[]>
45. Command Handler IncidentAddedEvent { SubscriptionId = 5,6,7 Incident: { … } } EventStream: IObservable<DomainEvent> WebSockets Middleware AMQP Middleware connections connections Dictionary<SubscriptionId, Connections[]> Dictionary<SubscriptionId, Connections[]>
46. Decoupled components via Pub/Sub messaging (IObservable<T>) publishEvent(event) CQRS (on micro scale)
47. Query Handler CQRS (on micro scale)
48. Query Handler Query -> Async<Result<Incident[], AppError>>
49. type QueryHandler(storage: InStorage) = let response (incidents: Incident[]) = incidents |> Array.map(Incident.toContractType) let handle (query) = asyncResult { match query with | Incident id -> match! storage.FindIncident(id) with | Some incident -> return response [| incident |] | None -> return response(Array.empty) } member x.Handle = handle
50. API Response CQRS (on micro scale)
51. Your Server as a function CQRS (on micro scale)
52. Your Server as a function HttpRequest -> HttpResponse
53. Your Server as a function HttpRequest -> Async<HttpResponse>
54. Your Server as a function HttpRequest -> Query -> QueryHandler -> Async<HttpResponse>
55. Your Server as a function HttpRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<HttpResponse>
56. HttpRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<HttpResponse> WebSocketRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<WebSocketResponse> AMQPRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<AMQPRequest>
57. HttpRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<HttpResponse> WebSocketRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<WebSocketResponse> AMQPRequest -> Query -> QueryHandler -> Async<Result<Incident[], AppError>> -> Async<AMQPRequest>
58. Error responses
59. type ParserError = | GameIdFormat of originValues:string[] | IncidentIdFormat of originValues:string[] | IncidentTypeFormat of originValues:string[] type DomainError = | IncidentCantBeCreated of gameId:int * typeId:int | LimitsOutOfRange of top:int * maxTop:int * skip:int | GameNotFound of gameId:int | IncidentNotFound of incidentId:int | IncidentsForGamesNotFound of gameIds:int[] type AppError = | Parser of ParserError | Domain of DomainError static member create(e: ParserError) = e |> Parser static member create(e: DomainError) = e |> Domain static member createResult(e: DomainError) = e |> Domain |> Error static member createResult(e: ParserError) = e |> Parser |> Error
60. module HTTPErrorResponse = let map (error: AppError) = ... let fromDomainError (error: DomainError) = match error with | IncidentCantBeCreated -> new HttpResponseMessage(HttpStatusCode.BadReqeust, { ... }) let fromParserError (error: ParserError) = match error with | GameIdFormat ids -> new HttpResponseMessage(HttpStatusCode.BadReqeust, { ... }) | IncidentIdFormat ids -> new HttpResponseMessage(HttpStatusCode.BadReqeust, { ... }) match error with | Domain e -> fromDomainError(e) | Parser e -> fromParserError(e)
61. Testing
62. Input -> Output Command -> Event
63. [<Fact>] let ``CommandHandler on AddIncident should return IncidentAdded if related game exist`` () = let cmdHandler = TestHelper.createCommandHandler() let game = TestHelper.createGame(id = 777) let incident = TestHelper.createIncident(id = 1, gameId = 777) asyncResult { let! gameEvent = Command.AddGame(game) |> cmdHandler.Handle let! incidentEvent = Command.AddIncident(incident) |> cmdHandler.Handle test <@ match incidentEvent with | IncidentAdded _ -> true | _ -> false @> } |> TestHelper.runTest
64. What about non-trivial cases with no output?
65. Testing Log Events
66. logger.Warning("Test message") logger.Warning("Test message") logger.Warning("Test message") InMemoryLogger .Should() .HaveMessage("Test message") .Appearing().Times(3) .WithLevel(LogEventLevel.Information)
67. [<Fact>] let ``Agent should be able to reconnect automatically and join the cluster`` () = async { let coordinatorDep = Dependency.createFor(NodeType.Coordinator) let (agentDep, loggerBuffer) = Dependency.createWithInMemoryLogger(NodeType.Agent) let server = MqttTests.startMqttServer() // we start one agent let! agent = Agent.init(agentDep, [| scenario |], agentSettings) Agent.startListening(agent) |> Async.Start // we stop mqtt server MqttTests.stopMqttServer server // and wait some time do! Async.Sleep(5_000) // spin up the mqtt server again to see that agent will reconnect let server = MqttTests.startMqttServer() let! coordinator = Coordinator.init(coordinatorDep, [| scenario |], coordinatorSettings, [| scenarioSettings |], customSettings) // waiting on reconnect do! Async.Sleep(2_000) let! statsResult = Coordinator.run(coordinator) let allStats = statsResult |> Result.defaultValue Array.empty
68. Property based testing
69. Property based testing [<Property>] let ``Subscription filter correctness verification`` (allTypes: IncidentType[], typesToSubscribe: IncidentType[]) = type IncidentType = Goal | Corner | Penalty | RedCard | YellowCard
70. Property based testing allTypes: [Goal; Corner; Penalty], typesToSubscribe: [Corner]) = type IncidentType = Goal | Corner | Penalty | RedCard | YellowCard
71. Property based testing allTypes: [Goal; Corner; Penalty; Corner;], typesToSubscribe: [RedCard; YellowCard]) = type IncidentType = Goal | Corner | Penalty | RedCard | YellowCard
72. [<Property>] let ``Subscription filter corectness verification`` (game: Game, allTypes: IncidentType list, typesToSubscribe: IncidentType list) = let cmdHandler = createCmdHandler() let incidents = allTypes |> List.map(fun x -> TestHelper.createIncidents(event) Command.AddGame(game) |> cmdHandler.Handle |> ignore Command.Subscribe([|game.Id|], typesToSubscribe) |> cmdHandler.Handle |> ignore let receivedIncidents = incidents |> List.map(fun x -> Command.AddIncident(x) |> cmdHandler.Handle |> retrieveIncidents |> List.sum let shouldReceive = if List.isEmpty(typesToSubscribe) // subscribed on all types then incidents.Length else let machedTypes = Set.intersect (set allTypes) (set typesToSubscribe) incidents |> List.choose(fun x -> if machedTypes.Contains(enum<IncidentType> x.IncidentTypeId) then Some 1 else None) |> List.sum receivedIncidents = shouldReceive
73. Part III Advantages and disadvantages
74. We stopped write frameworks with unlimited extensibility to cover cases from future
75. We stopped write frameworks No abstract factories No factory method
76. We stopped write frameworks No interfaces per class
77. We stopped write frameworks No interfaces like ISimpleIdentifiable
78. We stopped write frameworks Instead, we build our private world where we know everything
79. Simpler reading let publishMsg = Json.serialize >> Zip.compress >> RabbitMq.createMsg topicName >> client.Publish
80. Simpler reading let validate (context: NBomberContext) = context.Scenarios |> checkEmptyName >>= checkDuplicateName >>= checkEmptyStepName >>= checkDuration >>= checkConcurrentCopies >>= fun _ -> Ok context |> Result.mapError(AppError.create)
81. Simpler testing
82. Type Driven Development
83. module CardGame = type Suit = Club | Diamond | Spade | Heart type Rank = Two | Three | Four | Five | Six | Seven | Eight | Nine | Ten | Jack | Queen | King | Ace type Card = Suit * Rank type Hand = Card list type Deck = Card list type Player = { Name:string; Hand:Hand} type Game = {Deck:Deck; Players: Player list} type Deal = Deck -> (Deck*Card) type PickupCard = (Hand*Card) -> Hand
84. Anonymous Types let data = {| X = 1; Y = 2 |} let data1 = {| data with Y = 3 |} // {| X:Int; Y:Int |} let data2 = {| data with Z = 3 |} // {| X:Int; Y:Int; Z:Int |}
85. type User = { Age: int } let user = { Age = 1 } let anonymous = {| user with Y=1 |} // {| Age:Int; Y:Int |} Anonymous Types
86. Expressive API with Explicit Effects
87. Option<T> - to represent cases when something may exist but also may not. Result<TOk, TError> - to represent operations which could be success or fail. Choice<TLeft, TRight> - to model operations which return different values depending on the input Async<T> - to represent asynchronicity as an effect (in addition people quite often use F# Async<T> to model impure operations as alternative to IO monad)
88. Previously working in C# we were forced to use name conventions
89. // if method starts with Find then in case of not found // it returns null public User FindUser(int userId) // if method starts with Get then in case of not found // it throws Exception public User GetUser(int userId)
90. // if method starts with Try // it means that operation can fail and // and exception will be not thrown public bool TryParse(string json, out Config config) // in this case the exception will be thrown public Config Parse(string json)
91. let findUser (userId): Async<Result<Option<User>, AppError>> let parse (json): Result<Config, ParserError>
92. Monad Comprehensions
93. option { let! x = getSomething() let! y = getSomethingElse() let! z = andYetGetSomethingElse() // Code will only hit this point if the three // operations above return Some value return x + y + z }
94. type QueryHandler(storage: IStorage) = let handle (query) = asyncResult { match query with | Incidents (eventIds, incidentTypes, limits) -> // vLimits: Result<LimitsParams, AppError> let! vLimits = QueryValidation.validateLimits(limits) // incidents: Async<Result<Incident[], AppError>> let! incidents = storage.FindIncidents(eventIds, incidentTypes, vLimits) let data = incidents |> Array.map(Incident.toContractType) return { Data = data } } member x.Handle = handle
95. THANKS @antyadev antyadev@gmail.com