Leo Gorodinski eulerfx

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 resulting State for subsequent retrieval. One way to address this is by storing State is a key-value store. Another way is to use a SQL database. Yet another way is event sourcing. The benefit of even 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:

Both command sourcing (CS) and event sourcing (ES) rely on determinism for correctness.

The correctness condition for ES is the determinism (purity) of the function State -> Event -> State. Given that this function is deterministic, aka it always maps the same inputs to the same ouputs, we can rely on it to reconstitute state at any point in time. Determinism is typically achieved by ensuring that the Event has all required information to make the state transition, ie no side effects. The Event then is a sort of "closure" of all pertinent information about the event.

The correctness condition for CS is the determinism of the function State -> Command -> Event. Herein lies one of the distinctions between command sourcing and event sourcing - a program can control its output, but it not its input. Since one can't control the input, aka command, one can't in general, enrich it with all required information to make the above function deterministic. A consequence of this is that you can't simply replay a

There are some clear distinctions between commands and events but there are also some not so clear. I discuss the notion of a command as an interpretation of an event.

Commands are requests to do something, they are named in the imperative tense and capture intent. In CQRS, commands are distinguished from queries because they invoke behaviors therefore causing state changes whereas queries don't. In some frameworks, there is a distinction between commands and events in that one sends commands and publishes events. Commands are sent to a single logical handler whereas events are broadcast to potentially multiple handlers.

Events are characterized as being in the past tense. In addition to naming this imposes the constraint that events can't be rejected - you can't change the past. Furthermore, an event can be regarded as an outcome of a command - a causality relation.

There are perspectives where the distinction blurs. For example, what is the distinction between a command handler and an event handler?

	type Service<'Input, 'Output> = 'Input -> Async<'Output>

	/// A filter is an aspect in the AOP sense.
	type Filter<'Input, 'InputInner, 'OutputInner, 'Output> = 'Input-> Service<'InputInner, 'OutputInner> -> Async<'Output>

	type Filter<'Input, 'Output> = 'Input-> Service<'Input, 'Output> -> Async<'Output>

	type Continuation<'a,'r> = ('a -> 'r) -> 'r

	module Continuation =

	let subscribeAsAsyncSeq (conn:IEventStoreConnection) (stream:string) (resolveLinks:bool) (bufferSize:int) (ct:CancellationToken) = asyncSeq {
	use buffer = new System.Collections.Concurrent.BlockingCollection<ResolvedEvent>(bufferSize)
	let inline onEvent subs (evt:ResolvedEvent) =
	buffer.Add(evt)
	let inline onDrop (subs:EventStoreSubscription) (reason:SubscriptionDropReason) (ex:exn) =
	printfn "SUBSCRIPTION DROPPED! last position=%i reason=%O ex=%O" (subs.LastEventNumber.GetValueOrDefault()) reason ex
	()
	let! subs = conn.SubscribeToStreamAsync(stream, resolveLinks, Action<_,_>(onEvent), Action<_,_,_>(onDrop), null) \|> Async.AwaitTask
	yield! buffer.GetConsumingEnumerable(ct) \|> AsyncSeq.ofSeq
	}

	type MerchantSkuImported with
	static member Zero = { MerchantSkuImported.id = null ; merchant_id = null ; merchant_sku = null ; upc = null ; manufacturer = null ; brand = null ; part_no = null ; title = null ; description = null ; bullets = List.empty ; category = 0 ; category_path = List.empty ; multi_pack_quantity = None ; shipping_weight = None ; package_length = None ; package_width = None ; package_height = None ; main_image_url = None ; other_image_url = None ; price = None ; quantity = None ; start_selling_date = None }
	static member MerchantSku = (fun (x:MerchantSkuImported) -> x.merchant_sku) \|> Lens.create <\| fun v x -> { x with merchant_sku = v }
	static member UPC = (fun (x:MerchantSkuImported) -> x.upc) \|> Lens.create <\| fun v x -> { x with upc = v }
	static member Title = (fun (x:MerchantSkuImported) -> x.title) \|> Lens.create <\| fun v x -> { x with title = v }
	static member Brand = (fun (x:MerchantSkuImported) -> x.brand) \|> Lens.create <\| fun v x -> { x with brand = v }
	s

	type Event = {
	name : string
	date : DateTime
	code : int
	}

	let randEvent : Rand<Event> =
	let name = Rand.String (Rand.IntRange 10 15)
	let date = Rand.DateTime (DateTime(2014, 3, 26)) (DateTime(2015, 3, 26))
	let code = Rand.IntRange 1000 9999

	#load "HyperF.fsx"

	open HyperF
	open Route

	let service =
	[

	Get("/resource/:id") => fun (req,ri) -> "hello world!" \|> HttpRes.plainText

	/// For (1,"b",3) generates [1,"b",3]
	type TupleArrayConverter() =
	inherit JsonConverter()

	override x.CanConvert(typ) = FSharpType.IsTuple(typ)

	override x.WriteJson(writer, value, serializer) =
	let values = FSharpValue.GetTupleFields(value)
	serializer.Serialize(writer, values)

	module SeqStream

	/// Reads a specified number of values from an enumeration into a buffer array.
	let fillBuffer (enum:System.Collections.Generic.IEnumerator<_>) (buffer:_ array) offset count =
	let mutable read = 0
	while enum.MoveNext() && read < count do
	buffer.[read + offset] <- enum.Current
	read <- read + 1
	read