bartelink

Add consistent helpers for Async/Task/ValueTask`

F# has provided two complementary async mechanisms via FSharp.Core since V6: Async<'T> and Task<'T>. There's a long history of third party helper libraries providing shimming and/or alternate behaviors as expedient solutions to practical needs. For example, we have the long-awaited AwaitTaskCorrect resolution, and the fact that key Async helpers like bind, ignore, catch etc are (tupled) methods spread across the Async and AsyncBuilder types with PascalCased names (i.e. async.Bind vs Async.Ignore). Similarly, while there is an Async.Ignore, there is no task.Ignore or Task.ignore.

This suggestion proposes modules for Task, Async and ValueTask with consistent naming and pit-of-success behaviors OOTB via FSharp.Core.

Note this proposal is intentionally not seeking to be complete at the expense of drowning all F# users in a sea of new functions offering marginal gains; the aim is to cover 90% of common needs, with potential

bartelink

Events happen in an order. Why would you ever not just simply process them in that obvious order?

In the general case, most reaction and projection handling can/should actually be considered at the stream level. Why?

Flattening everything down into a denormalized SQL relational model complete with hierarchical foreign key constraints is the poster child case where you would benefit from a global order - if you traverse from start to finish in large batches and presume you're starting from a blank read model, you can map everything to large transactional batches of inserts and updates, and you save a checkpoint position as part of the same batch. If you want to pause the processing, you can simply resume from the checkpoint in your read model

In reality, life is not that simple/complicated:

• you may not be writing to a general database that you own, so you may not be able to have a transaction across your checkpoint and your model writes
• if you're feeding to a partner

bartelink

(*

let a = async {
    let! ct = Async.CancellationToken
    do! System.Threading.Tasks.Task.Delay(5000,ct) |> Async.AwaitTask
}

let b = async {
    failwith "B throw"
}

bartelink

/// Thread-safe coordinator that batches concurrent requests for a single <c>dispatch</> invocation such that:
/// - requests arriving together can be coalesced into the batch during the linger period via TryAdd
/// - callers that have had items admitted can concurrently await the shared fate of the dispatch via AwaitResult
/// - callers whose TryAdd has been denied can await the completion of the in-flight batch via AwaitCompletion
type internal AsyncBatch<'Req, 'Res>(dispatch : 'Req[] -> Async<'Res>, linger : System.TimeSpan) =
    let lingerMs = int linger.TotalMilliseconds
    // Yes, naive impl in the absence of a cleaner way to have callers sharing the AwaitCompletion coordinate the adding
    do if lingerMs < 5 then invalidArg "linger" "must be >= 5ms"
    let queue = new System.Collections.Concurrent.BlockingCollection<'Req>()
    let dispatch = async {

bartelink

### Keybase proof

I hereby claim:

  * I am bartelink on github.
  * I am bartelink (https://keybase.io/bartelink) on keybase.
  * I have a public key ASBsuiXja6_19Fq_ZG-VmuGchBYVYRFszqzRoRoy4vDl4go

To claim this, I am signing this object:

bartelink

namespace TickSpec

open System
open System.Reflection
open System.Threading.Tasks
open Xunit
open Xunit.Abstractions
open Xunit.Sdk

module Impl =

bartelink

A state machine is defined as follows:

• Input - a set of inputs
• Output - a set of outputs
• State - a set of states
• S0 ∈ S - an initial state
• T : Input * State -> Output * State - a transition function

If you model your services (aggregates, projections, process managers, sagas, whatever) as state machines, one issue to address is management of State. There must be a mechanism to provide State to the state machine, and to persist the resulting State for subsequent retrieval. One way to address this is by storing State in a key-value store. Another way is to use a SQL database. Yet another way is event sourcing. The benefit of event sourcing is that you never need to store State itself. Instead, you rely on the Output of a service to reconstitute state. In order to do that, the state machine transition function needs to be factored into two functions as follows:

bartelink

module FunDomain.Persistence.NEventStore.NesGateway

open FunDomain.Persistence.Serialization

open NEventStore
open NEventStore.Persistence
open NEventStore.Persistence.Sql.SqlDialects

open Microsoft.FSharp.Reflection
open System

bartelink

module WebApi2Helpers

// Port of answer by @nikosbaxevanis :- http://stackoverflow.com/a/19954215/11635

open System
open Ploeh.AutoFixture
open Ploeh.AutoFixture.Kernel
open Ploeh.AutoFixture.Xunit
open Ploeh.AutoFixture.AutoFoq
open System.Web.Http.Hosting

bartelink

type TypeThatIsARequestSpecification<'T>() =
    interface IRequestSpecification with
        member this.IsSatisfiedBy request = 
            match request with
            | :? System.Type as requestType -> typeof<'T>.IsAssignableFrom( requestType)
            | _ -> false
type PrepareSpecimenCommand<'T>( action) =
    interface ISpecimenCommand with
        member this.Execute( specimen, context) = 
            let instance = specimen :?> 'T