gregoryyoung

val currentState = events.foldLeft(0)(state, event) => event match {
         case d:InventoryItemDeactivated =>  Deactivated(d.Id, d.Reason)
         case c:InventoryItemCreated => Created(c.Id, c.Name)
         case a:InventoryItemAudited => Audited(a.Id, a.Outcome, a.EndValue)
}

//or extractors ....
//can put logic right there as well lambdas ...

gregoryyoung

Handlers

In Domain Driven Design, there are a series of small services that sit over the top of the domain. These services known as Application Services act as a facade over the the domain model. In CQRS based systems a similar pattern is used but it has been slightly refined. The refining of this pattern can be employed successfully in both systems that use CQRS and systems that stay using a single model for supporting reads and writes as many of the advantages come from simplicity of composition. To show Command Handlers, it is best to start with a typical Application Service and refactor our way to a Command Handler.

The Application Service

A stereotypical Application Service represents a given use case of the model. The Application Service itself generally does not directly implement the use case but instead coordinates objects from the domain to meet the requirements of the use case. This sounds like a small distinction but it is  a very important one, Application Services generally should not have "l

gregoryyoung

[05040,01,09:30:46.010] Processing command: WR.
[05040,06,09:30:46.010] Connected to [127.0.0.1:1113].
[05040,06,09:30:46.010] [127.0.0.1:1113]: Writing...
[05040,06,09:30:46.023] Successfully written. EventId: System.Byte[].
[05040,06,09:30:46.023] ##teamcity[buildStatisticValue key='WR-latency-ms' value='12']
[05040,06,09:30:46.023] Write request took: 00:00:00.0123514.
[05040,06,09:30:46.023] Connection [127.0.0.1:1113] was closed cleanly.
[05040,01,09:30:46.023] Command exited with code 0.
>>> WR

gregoryyoung

Here are some tests that show some of the failure modes of concurrent queue (all work in CLR impl)

Unhandled Exception:
System.NullReferenceException: Object reference not set to an instance of an object
  at System.Collections.Concurrent.ConcurrentQueue`1[T].TryDequeue (System.Collections.Concurrent.T& result) [0x00000] in <filename unknown>:0 
  at ConsoleApplication1.Program+<TestMonoConcurrentQueueReference>c__AnonStorey1.<>m__0 (System.Object x) [0x00000] in <filename unknown>:0 
  at System.Threading.Thread.StartInternal () [0x00000] in <filename unknown>:0 
[ERROR] FATAL UNHANDLED EXCEPTION: System.NullReferenceException: Object reference not set to an instance of an object
  at System.Collections.Concurrent.ConcurrentQueue`1[T].TryDequeue (System.Collections.Concurrent.T& result) [0x00000] in <filename unknown>:0 
  at ConsoleApplication1.Program+<TestMonoConcurrentQueueReference>c__AnonStorey1.<>m__0 (System.Object x) [0x00000] in <filename unknown>:0 

gregoryyoung

Micro Instance EC2


[02525,06,18:24:00.794] Completed. READS done: 10000.
[02525,06,18:24:00.794] 10000 requests completed in 10770ms (928.51 reqs per sec).

[02525,42,18:24:32.770] Write request took: 00:00:00.5000152.
[02525,42,18:24:32.773] Connection [127.0.0.1:1113] was closed cleanly.
[02525,01,18:24:32.773] Command exited with code 0.

gregoryyoung

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using EventStore.ClientAPI;

namespace marketdata
{

gregoryyoung

{
  "title": "Event stream 'ticks-AMX'",
  "id": "http://192.168.3.100:2113/streams/ticks-AMX",
  "updated": "2012-12-29T10:06:07.693857Z",
  "author": {
    "name": "EventStore"
  },
  "links": [
    {
      "uri": "http://192.168.3.100:2113/streams/ticks-AMX",

gregoryyoung

fromStream("$stats-127.0.0.1:2113").when( {
                        "$stats-collected" : function(s,e) { 
                            var total = e.body["sys-cpu"];
                            if(total > 40) {
                                emit("heavycpu", "heavycpufound", {});
                        }}
    })

gregoryyoung

{
  "proc-startTime": "2013-02-13T13:25:45Z",
  "proc-id": 6027,
  "proc-mem": 1092829184,
  "proc-cpu": 0.0,
  "proc-threadsCount": 0,
  "proc-contentionsRate": 0.0,
  "proc-thrownExceptionsRate": 0.0,
  "sys-cpu": 3.81117439,
  "sys-freeMem": 8884293632,

gregoryyoung

EventStore local node (serving full feeds, no reverse proxy)

greg@ouroboros:~/src/wrk$ ./wrk -c 20 -d 10 -t 10 http://127.0.0.1:2115/streams/newstream
Running 10s test @ http://127.0.0.1:2115/streams/newstream
  10 threads and 20 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     2.37ms    8.95ms 105.88ms   99.10%
    Req/Sec     0.00      0.00     0.00    100.00%
  22099 requests in 10.00s, 61.83MB read
  Socket errors: connect 0, read 0, write 0, timeout 44