February 2014 - Soloco

Event Store Projection in C#

2014-02-20/0 Comments/in Blog, EventStore /by Tim Cools

To process the irresponsible gambler alarms generated in our last post, we’ll be implementing an Event Store projection in C#. Why C# and not JavaScript, you ask? It’s true that this projection could be written in any language, but since there are plenty of C# projects and developers around the world, it only makes sense to demonstrate how to write an Event Store projection in C#.

Projection IrresponsibleGamblerAlarmPublisher

To retrieve the generated alarms, we’ll subscribe to all the events of the IrresponsibleGamblingAlarms stream. When we receive an alarm, we handle it by publishing it on a bus. After processing, we store the position of the last processed event in a checkpoint stream. These checkpoints allow us to continue from the last processed event in the event the projection is restarted or the connection to Event Store is dropped. In this example, we store a checkpoint every time an event is processed; however, if we need to reduce the load on Event Store, we can instead store checkpoints every, say, 1000 events.

In order to handle the generated alarm in this example, we will publish the generated alarms to a bus. This could be NServiceBus, some internal bus, or any other kind of bus we prefer. The advantage of this is that we can have multiple handlers for the same event.

C# implementation

public class IrresponsibleGamblerAlarmPublisher
{
    private const string CheckpointStream = "$publisher-IrresponsibleGamblerAlarmPublisher-checkpoint";
    private const string AlarmStream = "IrresponsibleGamblingAlarms";
         
    private readonly IEventStoreConnection _eventStoreConnection;
    private readonly IBus _bus;
    private readonly IConsole _console;

    private bool _running;

    public IrresponsibleGamblerAlarmPublisher(IEventStoreConnection eventStoreConnection, IBus bus, IConsole console)
    {
        if (eventStoreConnection == null) throw new ArgumentNullException("eventStoreConnection");
        if (bus == null) throw new ArgumentNullException("bus");
        if (console == null) throw new ArgumentNullException("console");

        _eventStoreConnection = eventStoreConnection;
        _bus = bus;
        _console = console;
    }

    public void Start()
    {
        if (_running) throw new InvalidOperationException("Projection already running");

        _running = true;

        Connect();
    }

    public void Stop()
    {
        if (!_running) throw new InvalidOperationException("Projection not running");

        _running = false;
    }

    private void Connect()
    {
        var position = GetLastCheckpoint(CheckpointStream);

        _eventStoreConnection.SubscribeToStreamFrom(AlarmStream, position, true, ProcessEvent,
            userCredentials: EventStoreCredentials.Default, subscriptionDropped: TryToReconnect);
    }

    private void TryToReconnect(EventStoreCatchUpSubscription catchUpSubscription, SubscriptionDropReason reason, Exception exception)
    {
        _console.Error("Projection subscription dropped: " + reason, exception);

        Connect();
    }   

    private void ProcessEvent(EventStoreCatchUpSubscription subscribtion, ResolvedEvent resolvedEvent)
    {
        var alarm = resolvedEvent.ParseJson<IrresponsibleGamblerDetected>();

        Publish(alarm);

        StoreCheckpoint(resolvedEvent);
    }

    private void Publish(IrresponsibleGamblerDetected alarm)
    {
        _bus.Publish(alarm);
    }

    private void StoreCheckpoint(ResolvedEvent resolvedEvent)
    {
        var eventNumber = resolvedEvent.Event.EventNumber;
        var checkpoint = new IrresponsibleGamblerAlarmPublisherCheckpoint(eventNumber)
            .AsJsonEvent();

        SetCheckpointStreamMaxCount(eventNumber);

        _eventStoreConnection.AppendToStream(CheckpointStream, ExpectedVersion.Any, EventStoreCredentials.Default, checkpoint);
    }

    private void SetCheckpointStreamMaxCount(int eventNumber)
    {
        if (eventNumber != 0) return;

        var metadata = StreamMetadata.Build().SetMaxCount(1);

        _eventStoreConnection.SetStreamMetadata(CheckpointStream, ExpectedVersion.Any, metadata, EventStoreCredentials.Default);
    }

    private int? GetLastCheckpoint(string stateStream)
    {
        var state = _eventStoreConnection.GetLastEvent<IrresponsibleGamblerAlarmPublisherCheckpoint>(stateStream);
        return state != null ? state.LastEventProcessed : (int?) null;
    }
}

public class IrresponsibleGamblerAlarmPublisher

{

private const string CheckpointStream = "$publisher-IrresponsibleGamblerAlarmPublisher-checkpoint";

private const string AlarmStream = "IrresponsibleGamblingAlarms";

private readonly IEventStoreConnection _eventStoreConnection;

private readonly IBus _bus;

private readonly IConsole _console;

private bool _running;

public IrresponsibleGamblerAlarmPublisher(IEventStoreConnection eventStoreConnection, IBus bus, IConsole console)

{

if (eventStoreConnection == null) throw new ArgumentNullException("eventStoreConnection");

if (bus == null) throw new ArgumentNullException("bus");

if (console == null) throw new ArgumentNullException("console");

_eventStoreConnection = eventStoreConnection;

_bus = bus;

_console = console;

}

public void Start()

{

if (_running) throw new InvalidOperationException("Projection already running");

_running = true;

Connect();

}

public void Stop()

{

if (!_running) throw new InvalidOperationException("Projection not running");

_running = false;

}

private void Connect()

{

var position = GetLastCheckpoint(CheckpointStream);

_eventStoreConnection.SubscribeToStreamFrom(AlarmStream, position, true, ProcessEvent,

userCredentials: EventStoreCredentials.Default, subscriptionDropped: TryToReconnect);

}

private void TryToReconnect(EventStoreCatchUpSubscription catchUpSubscription, SubscriptionDropReason reason, Exception exception)

{

_console.Error("Projection subscription dropped: " + reason, exception);

Connect();

}

private void ProcessEvent(EventStoreCatchUpSubscription subscribtion, ResolvedEvent resolvedEvent)

{

var alarm = resolvedEvent.ParseJson<IrresponsibleGamblerDetected>();

Publish(alarm);

StoreCheckpoint(resolvedEvent);

}

private void Publish(IrresponsibleGamblerDetected alarm)

{

_bus.Publish(alarm);

}

private void StoreCheckpoint(ResolvedEvent resolvedEvent)

{

var eventNumber = resolvedEvent.Event.EventNumber;

var checkpoint = new IrresponsibleGamblerAlarmPublisherCheckpoint(eventNumber)

.AsJsonEvent();

SetCheckpointStreamMaxCount(eventNumber);

_eventStoreConnection.AppendToStream(CheckpointStream, ExpectedVersion.Any, EventStoreCredentials.Default, checkpoint);

}

private void SetCheckpointStreamMaxCount(int eventNumber)

{

if (eventNumber != 0) return;

var metadata = StreamMetadata.Build().SetMaxCount(1);

_eventStoreConnection.SetStreamMetadata(CheckpointStream, ExpectedVersion.Any, metadata, EventStoreCredentials.Default);

}

private int? GetLastCheckpoint(string stateStream)

{

var state = _eventStoreConnection.GetLastEvent<IrresponsibleGamblerAlarmPublisherCheckpoint>(stateStream);

return state != null ? state.LastEventProcessed : (int?) null;

}

Starting the projection

When the projection starts, it reads the last position from the checkpoint stream, which is passed as a starting point to SubscribeToStreamFrom. If no checkpoint is found, null is passed instead, indicating that the subscription should be read from the start.

Handling the event

When an event is received, it is serialized to a typed irresponsiblegamblerdetected event and published on a (fake) bus. The position of the last handled event is then stored to the checkpoint stream. The first time a position is stored, the checkpoint stream’s metadata is set to $maxCount: 1. This indicates that we only need one event in the stream (i.e. the last one) and that the other events are safe to remove when the database is scavenged.

What if the projection crashes?

On the rare occasion our projection crashes after publishing to the bus but before storing a checkpoint, some events will end up published to the bus twice. Therefore, we can only guarantee that events are published at least once, and all our handlers should be idempotent or otherwise ready to handle the same events twice.

Concurrency

Because we will have only one projection writing to the CheckpointStream, and from a single thread at that, we can safely ignore concurrency checks when we write the projection using ExpectedVersion.Any.

Reconnecting

If the TCP connection drops, all we need to do to reconnect is call the Connect() method again. The projection will wait until the connection is reestablished, and then it will continue processing events as usual. This, of course, assumes that we’ve specified the KeepReconnecting connection setting when creating our Event Store connection. Otherwise, we won’t be able to reconnect at all if the connection drops.

var settings = ConnectionSettings.Create()
    .KeepReconnecting()
    .UseConsoleLogger();

var connection = EventStoreConnection.Create(settings, IPEndPointFactory.DefaultTcp());            
connection.Connect();

var settings = ConnectionSettings.Create()

.KeepReconnecting()

.UseConsoleLogger();

var connection = EventStoreConnection.Create(settings, IPEndPointFactory.DefaultTcp());

connection.Connect();

Source code

A working project for this example can be found on github: https://github.com/tim-cools/EventStore-Examples

EventStore Projections by Example

This post is part of a series:

Event Store Projections – Temporal projection for generating alarms

2014-02-16/0 Comments/in Blog, EventStore /by Tim Cools

To detect whether a player is gambling irresponsibly, we total the amounts won and lost over all games played in the last 24 hours. If this amount exceeds 500€, we generate an alarm – but since we don’t want to spam our players, we only do so if the previous alarm was more than 24 hours ago.

Projection IrresponsibleGamblingDetector

The projection takes all GameWon and GameLost events from all Player streams as input and stores all games from the last 24 hours in its state. When the amount of money a player loses exceeds 500€ in a 24-hour span, it generates an alarm by emitting an event to the IrresponsibleGambleAlarm stream.

Projection IrresponsibleGamblingDetector-Events

JavaScript Implementation

The irresponsibleGamblingDetector module implements the core projection logic:

// <reference path="References\1Prelude.js"></reference>
// <reference path="References\Projections.js"></reference>

var irresponsibleGamblingDetector = function irresponsibleGamblingDetectorConstuctor($eventServices) {

    var eventServices = !$eventServices ? { emit: emit } : $eventServices;
   
    var millisecondsPerDay = 24 * 60 * 60 * 1000;
    var amountLostThreshold = 500;
    
    var emitAlarm = function (state, playerId, total, timestamp)  {

        state.LastAlarm = timestamp;

        eventServices.emit('IrresponsibleGamblingAlarms', 'IrresponsibleGamblerDetected', {
            PlayerId: playerId,
            AmountSpentLAst24Hours: total,
            Timestamp: timestamp
        });
    };
    
    var init = function () {
        return {            
            LastAlarm: null,
            GamesLast24Hour: [],
        };
    };

    var isMoreAs24HoursDifference = function (timestamp, reference) {

        if (reference === null) return true;

        var millisecondsDifference = new Date(timestamp).getTime() - new Date(reference).getTime();

        return millisecondsDifference > millisecondsPerDay  ;
    };

    var removeGamesOlderThan24HoursFromCache = function (gamesResults, timestamp) {

        while (gamesResults.length > 0 && isMoreAs24HoursDifference(timestamp, gamesResults[0].Timestamp)) {
            gamesResults.pop(0);
        }
    };
    
    var addNewGameToCache = function (gamesResults, timestamp, amount, gameId) {

        gamesResults.push({ Timestamp: timestamp, Amount: amount, GameId: gameId });
    };
    
    var updateCachedGamesLast24Hour = function (gamesResults, timestamp, amount, gameId) {

        removeGamesOlderThan24HoursFromCache(gamesResults, timestamp);
        addNewGameToCache(gamesResults, timestamp, amount, gameId);
    };

    var calculateTotalAmount = function (gamesResults) {

        var total = 0;
        for (var resultIndex = 0; resultIndex < gamesResults.length; resultIndex++) {
            var result = gamesResults[resultIndex];
            total += result.Amount;
        }
        return total;
    };

    var duplicated = function (gamesResults, gameId) {
        
        for (var resultIndex = 0; resultIndex < gamesResults.length; resultIndex++) {

            var result = gamesResults[resultIndex];
            if (result.GameId == gameId) {
                return true;
            }
        }
        
        return false;
    };

    var process = function (state, playerId, timestamp, amount, gameId) {
        
        var gamesResults = state.GamesLast24Hour;

        if (duplicated(gamesResults, gameId)) {
            return state;
        }

        updateCachedGamesLast24Hour(gamesResults, timestamp, amount, gameId);

        if (!isMoreAs24HoursDifference(timestamp, state.LastAlarm)) {
            return state;
        }

        var total = calculateTotalAmount(gamesResults);        
        if (total < -amountLostThreshold) {
            emitAlarm(state, playerId, total, timestamp);
        }
                
        return state;
    };

    var processEvent = function (state, event) {

        var playerId = event.body.PlayerId;
        var timestamp = event.body.Timestamp;
        var amount = event.body.Amount;
        var gameId = event.body.GameId;
        
        return process(state, playerId, timestamp, amount, gameId);
    };

    return {
        init: init,
        processGameLost: processEvent,
        processGameWon: processEvent
    };
};

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

// <reference path="References\1Prelude.js"></reference>

// <reference path="References\Projections.js"></reference>

var irresponsibleGamblingDetector = function irresponsibleGamblingDetectorConstuctor($eventServices) {

var eventServices = !$eventServices ? { emit: emit } : $eventServices;

var millisecondsPerDay = 24 * 60 * 60 * 1000;

var amountLostThreshold = 500;

var emitAlarm = function (state, playerId, total, timestamp) {

state.LastAlarm = timestamp;

eventServices.emit('IrresponsibleGamblingAlarms', 'IrresponsibleGamblerDetected', {

PlayerId: playerId,

AmountSpentLAst24Hours: total,

Timestamp: timestamp

});

};

var init = function () {

return {

LastAlarm: null,

GamesLast24Hour: [],

};

var isMoreAs24HoursDifference = function (timestamp, reference) {

if (reference === null) return true;

var millisecondsDifference = new Date(timestamp).getTime() - new Date(reference).getTime();

return millisecondsDifference > millisecondsPerDay ;

};

var removeGamesOlderThan24HoursFromCache = function (gamesResults, timestamp) {

while (gamesResults.length > 0 && isMoreAs24HoursDifference(timestamp, gamesResults[0].Timestamp)) {

gamesResults.pop(0);

}

};

var addNewGameToCache = function (gamesResults, timestamp, amount, gameId) {

gamesResults.push({ Timestamp: timestamp, Amount: amount, GameId: gameId });

};

var updateCachedGamesLast24Hour = function (gamesResults, timestamp, amount, gameId) {

removeGamesOlderThan24HoursFromCache(gamesResults, timestamp);

addNewGameToCache(gamesResults, timestamp, amount, gameId);

};

var calculateTotalAmount = function (gamesResults) {

var total = 0;

for (var resultIndex = 0; resultIndex < gamesResults.length; resultIndex++) {

var result = gamesResults[resultIndex];

total += result.Amount;

}

return total;

};

var duplicated = function (gamesResults, gameId) {

for (var resultIndex = 0; resultIndex < gamesResults.length; resultIndex++) {

var result = gamesResults[resultIndex];

if (result.GameId == gameId) {

return true;

}

return false;

};

var process = function (state, playerId, timestamp, amount, gameId) {

var gamesResults = state.GamesLast24Hour;

if (duplicated(gamesResults, gameId)) {

return state;

}

updateCachedGamesLast24Hour(gamesResults, timestamp, amount, gameId);

if (!isMoreAs24HoursDifference(timestamp, state.LastAlarm)) {

return state;

}

var total = calculateTotalAmount(gamesResults);

if (total < -amountLostThreshold) {

emitAlarm(state, playerId, total, timestamp);

}

return state;

};

var processEvent = function (state, event) {

var playerId = event.body.PlayerId;

var timestamp = event.body.Timestamp;

var amount = event.body.Amount;

var gameId = event.body.GameId;

return process(state, playerId, timestamp, amount, gameId);

};

return {

init: init,

processGameLost: processEvent,

processGameWon: processEvent

};

The init method generates the initial state, which contains the timestamp of the last generated alarm and an empty list of the games that player has played in the last 24 hours.

The processGameLost and processGameWon methods process the corresponding events. First, they update the list of games played in the last 24 hours, removing games that were played more than 24 hours ago and adding newly received games. Next, they check whether an alarm was generated in the last 24 hours: if this is the case, processing halts immediately. Finally, they calculate the total amount won or lost in all games over the last 24 hours, and if the amount lost exceeds the threshold, then an IrresponsibleGamblerDetected event is emitted to the IrresponsibleGamblingAlarms stream.

Since we can only guarantee that events are processed at least once, we also have to ensure that our projection is idempotent. We need to check whether an incoming event has already been processed, or else we’ll get strange results if we process the results of the same game twice (or more). By storing the GameIds of each game we process in the state, we can determine whether we’ve already seen a specific game or not before we process it.

This module is instantiated once globally and then referenced from the projection’s definition:

var detector = irresponsibleGamblingDetector();

fromCategory('Player')
    .foreachStream()
    .when({
        $init: detector.init,
        GameLost: detector.processGameLost,
        GameWon: detector.processGameWon
    });

var detector = irresponsibleGamblingDetector();

fromCategory('Player')

.foreachStream()

.when({

$init: detector.init,

GameLost: detector.processGameLost,

GameWon: detector.processGameWon

});

This definition indicates that the projection is only interested in GameLost and GameWon events from Player streams.

Testing the projection

To test the projection, we’ll define four scenarios. The first will test a lost game, but without exceeding the threshold, so no alarm will be generated. The second will assert that an alarm is generated when the threshold is exceeded. The third scenario will also exceed the threshold, but in more than 24 hours, so no alarm will be generated. The last scenario will check for event duplication.

/// <reference path="../References/jasmine/jasmine.js"></reference>
/// <reference path="../IrresponsibleGamblingDetector.js"></reference>

describe("when detecting irresponsible gamblers", function () {

    var detector;
    var projections;

    beforeEach(function () {
        projections = jasmine.createSpyObj('projections', ['emit']);
        detector = irresponsibleGamblingDetector(projections);
    });

    describe("when processing GameLost event", function () {

        describe("given amount lost is less than 500", function () {

            var state;

            beforeEach(function () {

                state = detector.init();
                state = detector.processGameLost(state, {
                    sequenceNumber: 10,
                    body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }
                });
            });
            
            it("state should contain the one amount", function () {

                expect(state).toEqual({
                    LastAlarm: null,
                    GamesLast24Hour: [
                        { Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' }
                    ]
                });
            });

            it("no event should be emitted", function () {

                expect(projections.emit).wasNotCalled();
            });
        });

        describe("given amount lost is greater than 500 in last 24 hours", function () {

            var state;

            beforeEach(function () {

                state = detector.init();
                state = detector.processGameLost(state, {
                    sequenceNumber: 1,
                    body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }
                });

                state = detector.processGameLost(state, {
                    sequenceNumber: 2,
                    body: { PlayerId: 'Player-1', GameId: 'Game-2', Amount: -401, Timestamp: '2013-12-19T08:02:39.687Z' }
                });
            });

            it("state should contain the both amounts", function () {

                expect(state).toEqual({
                    LastAlarm: '2013-12-19T08:02:39.687Z',
                    GamesLast24Hour: [
                        { Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' },
                        { Timestamp: '2013-12-19T08:02:39.687Z', Amount: -401, GameId: 'Game-2' }
                    ]
                });
            });

            it("should emit an alarm", function () {

                expect(projections.emit).wasCalledWith("IrresponsibleGamblingAlarms", "IrresponsibleGamblerDetected",
                    { PlayerId: 'Player-1', AmountSpentLast24Hours: -501, Timestamp: '2013-12-19T08:02:39.687Z' }
                );
            });
        });

        describe("given amount lost is greater than 500 but not in the last 24 hours", function () {

            var state;

            beforeEach(function () {

                state = detector.init();
                state = detector.processGameLost(state, {
                    body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }
                });

                state = detector.processGameLost(state, {
                    body: { PlayerId: 'Player-1', GameId: 'Game-2', Amount: -401, Timestamp: '2013-12-19T09:02:39.687Z' }
                });
            });

            it("state should contain the only last amount", function () {

                expect(state).toEqual({
                    LastAlarm: null,
                    GamesLast24Hour: [
                        { Timestamp: '2013-12-19T09:02:39.687Z', Amount: -401,  GameId: 'Game-2' }
                    ]
                });
            });

            it("should not emit an alarm", function () {

                expect(projections.emit).wasNotCalled();
            });
        });

        describe("given an event has already been processed", function () {

            var state;

            beforeEach(function () {

                var event = {
                    body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }
                };
                
                state = detector.init();
                state = detector.processGameLost(state, event);
                state = detector.processGameLost(state, event);
                state = detector.processGameLost(state, event);
            });

            it("state should contain the only one event", function () {

                expect(state).toEqual({
                    LastAlarm: null,
                    GamesLast24Hour: [
                        { Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' }
                    ]
                });
            });
        });
    });
});

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

/// <reference path="../References/jasmine/jasmine.js"></reference>

/// <reference path="../IrresponsibleGamblingDetector.js"></reference>

describe("when detecting irresponsible gamblers", function () {

var detector;

var projections;

beforeEach(function () {

projections = jasmine.createSpyObj('projections', ['emit']);

detector = irresponsibleGamblingDetector(projections);

});

describe("when processing GameLost event", function () {

describe("given amount lost is less than 500", function () {

var state;

beforeEach(function () {

state = detector.init();

state = detector.processGameLost(state, {

sequenceNumber: 10,

body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }

});

it("state should contain the one amount", function () {

expect(state).toEqual({

LastAlarm: null,

GamesLast24Hour: [

{ Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' }

]

});

it("no event should be emitted", function () {

expect(projections.emit).wasNotCalled();

});

describe("given amount lost is greater than 500 in last 24 hours", function () {

var state;

beforeEach(function () {

state = detector.init();

state = detector.processGameLost(state, {

sequenceNumber: 1,

body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }

});

state = detector.processGameLost(state, {

sequenceNumber: 2,

body: { PlayerId: 'Player-1', GameId: 'Game-2', Amount: -401, Timestamp: '2013-12-19T08:02:39.687Z' }

});

it("state should contain the both amounts", function () {

expect(state).toEqual({

LastAlarm: '2013-12-19T08:02:39.687Z',

GamesLast24Hour: [

{ Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' },

{ Timestamp: '2013-12-19T08:02:39.687Z', Amount: -401, GameId: 'Game-2' }

]

});

it("should emit an alarm", function () {

expect(projections.emit).wasCalledWith("IrresponsibleGamblingAlarms", "IrresponsibleGamblerDetected",

{ PlayerId: 'Player-1', AmountSpentLast24Hours: -501, Timestamp: '2013-12-19T08:02:39.687Z' }

);

});

describe("given amount lost is greater than 500 but not in the last 24 hours", function () {

var state;

beforeEach(function () {

state = detector.init();

state = detector.processGameLost(state, {

body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }

});

state = detector.processGameLost(state, {

body: { PlayerId: 'Player-1', GameId: 'Game-2', Amount: -401, Timestamp: '2013-12-19T09:02:39.687Z' }

});

it("state should contain the only last amount", function () {

expect(state).toEqual({

LastAlarm: null,

GamesLast24Hour: [

{ Timestamp: '2013-12-19T09:02:39.687Z', Amount: -401, GameId: 'Game-2' }

]

});

it("should not emit an alarm", function () {

expect(projections.emit).wasNotCalled();

});

describe("given an event has already been processed", function () {

var state;

beforeEach(function () {

var event = {

body: { PlayerId: 'Player-1', GameId: 'Game-1', Amount: -100, Timestamp: '2013-12-18T08:02:39.687Z' }

};

state = detector.init();

state = detector.processGameLost(state, event);

});

it("state should contain the only one event", function () {

expect(state).toEqual({

LastAlarm: null,

GamesLast24Hour: [

{ Timestamp: '2013-12-18T08:02:39.687Z', Amount: -100, GameId: 'Game-1' }

]

});

Once again, we see the power of event processing testing. We can arrange messages in the past and future without any issue.

Processing the generated alarms

Since we want to send email and text messages based on the output of this projection, we’ll be implementing a projection in C# to handle the alarms.

Source code

A working project for this example can be found on github: https://github.com/tim-cools/EventStore-Examples

Event Store Projections by Example

This post is part of a series:

Event Store Projections – Distributing events to other streams

2014-02-08/0 Comments/in Blog /by Tim Cools

Since the GameOver events are appended to the Game stream, we need a way to distribute them to all participating players’ streams. The GameOver event contains a list of participating players, and the distributor emits an event for each player. The event type emitted depends on how much the player has won or lost: if the amount is less than zero, a GameLost event is emitted, but if the amount is greater than zero, a GameWon event is emitted instead. (For simplicity’s sake, we’ll assume GameDraw events don’t exist.)

Note that we don’t technically have to split the event in two to implement our alarm system, but it’s useful for the purposes of this example.

Projection GameOverToPlayerDistributor

This projection takes GameOver events from all Game streams as input. For each GameOver event, it will loop over the list of players and distribute (i.e. emit) GameLost and GameWon events to the players’ streams as appropriate.

Projection-GameOverToPlayerDistributor-Events

JavaScript implementation

This projection module has a single public method, process, which processes the GameOver events by looping over the participating players and emitting the appropriate events:

// <reference path="References\1Prelude.js"></reference>
// <reference path="References\Projections.js"></reference>

var gameOverToPlayerDistributor = function gameOverToPlayerDistributorConstuctor($eventServices) {

    var eventServices = !$eventServices ? { emit: emit } : $eventServices;

    var createEvent = function (playerId, gameId, amount, timestamp) {
        return {
            PlayerId: playerId,
            GameId: gameId,
            Amount: amount,
            Timestamp: timestamp
        };
    };

    var emitGameEvent = function (eventType, gameId, playerId, amount, timestamp) {

        var event = createEvent(playerId, gameId, amount, timestamp);

        eventServices.emit(playerId, eventType, event);
    };

    var processPlayer = function (gameId, playerResult, timestamp) {
        if (playerResult.Amount > 0) {
            emitGameEvent("GameWon", gameId, playerResult.PlayerId, playerResult.Amount, timestamp);
        } else if (playerResult.Amount < 0) {
            emitGameEvent("GameLost", gameId, playerResult.PlayerId, playerResult.Amount, timestamp);
        }
    };
    
    var process = function (state, event) {

        var gameId = event.body.GameId;
        var players = event.body.PlayerResults;
        var timestamp = event.body.Timestamp;

        for (var playerIndex = 0; playerIndex < players.length; playerIndex++) {
            var player = players[playerIndex];
            processPlayer(gameId, player, timestamp);
        }
    };

    return {
        process: process
    };
};

// <reference path="References\1Prelude.js"></reference>

// <reference path="References\Projections.js"></reference>

var gameOverToPlayerDistributor = function gameOverToPlayerDistributorConstuctor($eventServices) {

var eventServices = !$eventServices ? { emit: emit } : $eventServices;

var createEvent = function (playerId, gameId, amount, timestamp) {

return {

PlayerId: playerId,

GameId: gameId,

Amount: amount,

Timestamp: timestamp

};

var emitGameEvent = function (eventType, gameId, playerId, amount, timestamp) {

var event = createEvent(playerId, gameId, amount, timestamp);

eventServices.emit(playerId, eventType, event);

};

var processPlayer = function (gameId, playerResult, timestamp) {

if (playerResult.Amount > 0) {

emitGameEvent("GameWon", gameId, playerResult.PlayerId, playerResult.Amount, timestamp);

} else if (playerResult.Amount < 0) {

emitGameEvent("GameLost", gameId, playerResult.PlayerId, playerResult.Amount, timestamp);

}

};

var process = function (state, event) {

var gameId = event.body.GameId;

var players = event.body.PlayerResults;

var timestamp = event.body.Timestamp;

for (var playerIndex = 0; playerIndex < players.length; playerIndex++) {

var player = players[playerIndex];

processPlayer(gameId, player, timestamp);

}

};

return {

process: process

};

This module is instantiated once globally and then referenced from the projection’s definition:

var distributor = gameOverToPlayerDistributor();

fromCategory('Game')
    .when({
        GameOver: distributor.process
    });

var distributor = gameOverToPlayerDistributor();

fromCategory('Game')

.when({

GameOver: distributor.process

});

Testing the projection

To test this projection, we’ll define three scenarios. The first will arrange an event with a positive amount and assert that a GameWon event is emitted. The second will arrange an event with a negative amount and assert that a GameLost event is emitted. The last will check for a combination of both.

/// <reference path="../References/jasmine/jasmine.js"></reference>
/// <reference path="../GameOverToPlayerDistributor.js"></reference>

describe("when distributing GameOver event to to players", function () {

    var distributor;
    var projections;

    beforeEach(function () {
        projections = jasmine.createSpyObj('projections', ['emit']);
        distributor = gameOverToPlayerDistributor(projections);
    });

    describe("when processing GameOver event", function () {

        describe("given amount is positive", function () {

            it("should emit a GameWon event to the player stream", function () {

                distributor.process({}, {
                    body: {
                        GameId: 'Game-abc',
                        Timestamp: '2014-02-20T08:02:39.687Z',
                        PlayerResults: [
                            { PlayerId: 'Player-p1', Amount: 100 }
                        ]
                    }
                });

                expect(projections.emit)
                   .wasCalledWith("Player-p1", "GameWon", {
                       PlayerId: 'Player-p1',
                       GameId: 'Game-abc',
                       Amount: 100,
                       Timestamp: '2014-02-20T08:02:39.687Z'
                   });
            });
        });

        describe("given amount is negative", function () {

            it("should emit a GameLost event to the player stream", function () {

                distributor.process({}, {
                    body: {
                        GameId: 'Game-abc',
                        Timestamp: '2014-02-20T08:02:39.687Z',
                        PlayerResults: [
                            { PlayerId: 'Player-p1', Amount: -100 }
                        ]
                    }
                });

                expect(projections.emit)
                    .wasCalledWith("Player-p1", "GameLost", {
                        PlayerId: 'Player-p1',
                        GameId: 'Game-abc',
                        Amount: -100,
                        Timestamp: '2014-02-20T08:02:39.687Z'
                    });
            });
        });


        describe("given multiple players", function () {

            beforeEach(function () {
                distributor.process({}, {
                    body: {
                        GameId: 'Game-abc',
                        Timestamp: '2014-02-20T08:02:39.687Z',
                        PlayerResults: [
                            { PlayerId: 'Player-p1', Amount: -100 },
                            { PlayerId: 'Player-p2', Amount: 80 },
                            { PlayerId: 'Player-p3', Amount: -40 },
                            { PlayerId: 'Player-p4', Amount: 20 }
                        ]
                    }
                });
            });

            it("should emit a GameWon event to the winning players stream", function () {

                expect(projections.emit)
                    .wasCalledWith("Player-p2", "GameWon", {
                        PlayerId: 'Player-p2',
                        GameId: 'Game-abc',
                        Amount: 80,
                        Timestamp: '2014-02-20T08:02:39.687Z',
                    });

                expect(projections.emit)
                    .wasCalledWith("Player-p4", "GameWon", {
                        PlayerId: 'Player-p4',
                        GameId: 'Game-abc',
                        Amount: 20,
                        Timestamp: '2014-02-20T08:02:39.687Z',
                    });
            });

            it("should emit a GameLost event to the losing players stream", function () {

                expect(projections.emit)
                    .wasCalledWith("Player-p1", "GameLost", {
                        PlayerId: 'Player-p1',
                        GameId: 'Game-abc',
                        Amount: -100,
                        Timestamp: '2014-02-20T08:02:39.687Z',
                    });

                expect(projections.emit)
                    .wasCalledWith("Player-p3", "GameLost", {
                        PlayerId: 'Player-p3',
                        GameId: 'Game-abc',
                        Amount: -40,
                        Timestamp: '2014-02-20T08:02:39.687Z',
                    });
            });
        });
    });
});

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

/// <reference path="../References/jasmine/jasmine.js"></reference>

/// <reference path="../GameOverToPlayerDistributor.js"></reference>

describe("when distributing GameOver event to to players", function () {

var distributor;

var projections;

beforeEach(function () {

projections = jasmine.createSpyObj('projections', ['emit']);

distributor = gameOverToPlayerDistributor(projections);

});

describe("when processing GameOver event", function () {

describe("given amount is positive", function () {

it("should emit a GameWon event to the player stream", function () {

distributor.process({}, {

body: {

GameId: 'Game-abc',

Timestamp: '2014-02-20T08:02:39.687Z',

PlayerResults: [

{ PlayerId: 'Player-p1', Amount: 100 }

]

}

});

expect(projections.emit)

.wasCalledWith("Player-p1", "GameWon", {

PlayerId: 'Player-p1',

GameId: 'Game-abc',

Amount: 100,

Timestamp: '2014-02-20T08:02:39.687Z'

});

describe("given amount is negative", function () {

it("should emit a GameLost event to the player stream", function () {

distributor.process({}, {

body: {

GameId: 'Game-abc',

Timestamp: '2014-02-20T08:02:39.687Z',

PlayerResults: [

{ PlayerId: 'Player-p1', Amount: -100 }

]

}

});

expect(projections.emit)

.wasCalledWith("Player-p1", "GameLost", {

PlayerId: 'Player-p1',

GameId: 'Game-abc',

Amount: -100,

Timestamp: '2014-02-20T08:02:39.687Z'

});

describe("given multiple players", function () {

beforeEach(function () {

distributor.process({}, {

body: {

GameId: 'Game-abc',

Timestamp: '2014-02-20T08:02:39.687Z',

PlayerResults: [

{ PlayerId: 'Player-p1', Amount: -100 },

{ PlayerId: 'Player-p2', Amount: 80 },

{ PlayerId: 'Player-p3', Amount: -40 },

{ PlayerId: 'Player-p4', Amount: 20 }

]

}

});

it("should emit a GameWon event to the winning players stream", function () {

expect(projections.emit)

.wasCalledWith("Player-p2", "GameWon", {

PlayerId: 'Player-p2',

GameId: 'Game-abc',

Amount: 80,

Timestamp: '2014-02-20T08:02:39.687Z',

});

expect(projections.emit)

.wasCalledWith("Player-p4", "GameWon", {

PlayerId: 'Player-p4',

GameId: 'Game-abc',

Amount: 20,

Timestamp: '2014-02-20T08:02:39.687Z',

});

it("should emit a GameLost event to the losing players stream", function () {

expect(projections.emit)

.wasCalledWith("Player-p1", "GameLost", {

PlayerId: 'Player-p1',

GameId: 'Game-abc',

Amount: -100,

Timestamp: '2014-02-20T08:02:39.687Z',

});

expect(projections.emit)

.wasCalledWith("Player-p3", "GameLost", {

PlayerId: 'Player-p3',

GameId: 'Game-abc',

Amount: -40,

Timestamp: '2014-02-20T08:02:39.687Z',

});

Querying the results with the Client API

The results of this projection are used solely as input for another stream, so there’s no need to read them with the C# Client API.

Source code

A working project for this example can be found on github: https://github.com/tim-cools/EventStore-Examples

Event Store Projections by Example

This post is part of a series:

Event Store Projections – The irresponsible gambler

2014-02-08/0 Comments/in Blog, EventStore /by Tim Cools

Imagine a world where online gambling sites were nice enough to notify you when you’re gambling irresponsibly: if you’d spend too much money, they’d take notice and send you an alert via email or text message. Of course, online gambling sites bank on people being just so frivolous with their money, but there’s nothing stopping us from dreaming – or from implementing such a system as our next example.

Image Source

First off, we’ll assume we have a Game aggregate which raises a GameOver that contains a list of the players that participated in the game and the amounts they lost or won.

Our irresponsible gambler alarm system needs three projections. The first projection will distribute the GameOver events to GameWon and GameLost events emitted to all participating players. The second projection is a temporal projection that will generate alarms based on the results of all games played in the last 24 hours. The last projection, implemented in C#, will send emails and text messages to any players who lost more than 500€ in the last 24 hours, our eponymous irresponsible gamblers. To process these alarms, we will publish all alarms generated in the second projection on a bus in C#.

You can find the projections here:

EventStore Projections by Example

This post is part of a series:

Event Store Projection in C#

C# implementation

Starting the projection

Handling the event

What if the projection crashes?

Concurrency

Reconnecting

Source code

EventStore Projections by Example

Event Store Projections – Temporal projection for generating alarms

JavaScript Implementation

Testing the projection

Processing the generated alarms

Source code

Event Store Projections by Example

Event Store Projections – Distributing events to other streams

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

Event Store Projections by Example

Event Store Projections – The irresponsible gambler

EventStore Projections by Example

Base camp

Pages

Archive for month: February, 2014

C# implementation

Starting the projection

Handling the event

What if the projection crashes?

Concurrency

Reconnecting

Source code

EventStore Projections by Example

JavaScript Implementation

Testing the projection

Processing the generated alarms

Source code

Event Store Projections by Example

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

Event Store Projections by Example

EventStore Projections by Example

Base camp

Blog – Tags

Pages