Archive for month: January, 2014

EventStore Projections – Calculating daily averages

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

This projection will calculate the daily average consumption per device. We take our Device streams as input and store an intermediate calculation in the state for each one. This calculation contains the total sum of measurements for the current day. The projection’s result is a new stream for each device which contains a new calculated average for each day.

The $init method initializes the state. Each subsequent MeasurementRead event will invoke the calculate method, which will update the state with the total number of measurements, the sum of the measurements, and the timestamp of the last measurement. When we determine from the timestamp that a full day has passed, we emit an event which contains the date, the sum, and the calculated average, obtained by dividing the sum of the measurements by the number of measurements.

JavaScript implementation

First, we implement a daily timestamp formatter which will handle formatting the date part of a JavaScript Date object. Normally, we’d use an existing library such as momentjs to format the date part, but Event Store doesn’t currently let us include external libraries. One workaround for this would be to inject the library in the projection source when the projection is created. I might cover this option in a future blog post, but for now, we’ll have to roll our own function.

// 
// 

var dailyTimestampFormatter = function dailyTimestampFormatterConstructor() {

    var zeroPad = function (num, places) {
        var zero = places - num.toString().length + 1;
        return Array(+(zero > 0 && zero)).join("0") + num;
    };

    var format = function (timestamp) {
        var date = new Date(timestamp);
        return date.getFullYear().toString() + '/' +
            zeroPad(date.getMonth() + 1, 2) + '/' +
            zeroPad(date.getDate(), 2);
    };
    
    return {
        format: format
    };
}();

var dailyTimestampFormatter = function dailyTimestampFormatterConstructor() {

var zeroPad = function (num, places) {

var zero = places - num.toString().length + 1;

return Array(+(zero > 0 && zero)).join("0") + num;

};

var format = function (timestamp) {

var date = new Date(timestamp);

return date.getFullYear().toString() + '/' +

zeroPad(date.getMonth() + 1, 2) + '/' +

zeroPad(date.getDate(), 2);

};

return {

format: format

};

}();

Next, we implement the projection module with the init and the cacluate methods:

var measurementReadAveragPerDayCalculator = function measurementReadAveragPerDayCalculatorConstuctor($eventServices) {

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

    var isFirstTime = function(previousState) {
         return !previousState.lastTimestamp;
    };

    var isSameTimeSlot = function (currentTimestamp, lastTimestamp) {
        return dailyTimestampFormatter.format(currentTimestamp)
            == dailyTimestampFormatter.format(lastTimestamp);
    };

    var createEvent = function (timeStamp, reading, count) {
        return {
            Timeslot: dailyTimestampFormatter.format(timeStamp),
            Total: reading,
            Count: count,
            Average: reading / count
        };
    };

    var createState = function (reading, count, timestamp) {
        return {
            total: reading,
            count: count,
            average: reading / count,
            lastTimestamp: timestamp
        };
    };

    var emitAverageEvent = function (measurementEvent, previousState) {

        var streamName = 'MeasurementAverageDay-' + measurementEvent.streamId;
        var event = createEvent(previousState.lastTimestamp, previousState.total, previousState.count);

        eventServices.emit(streamName, "MeasurementAverageDay", event);
    };

    var init = function () {
        return createState(0, 0, null);
    };
    
    var update = function (previousState, measurementEvent) {

        var timestamp = measurementEvent.body.Timestamp;
        var reading = measurementEvent.body.Reading;

        if (isFirstTime(previousState) || isSameTimeSlot(timestamp, previousState.lastTimestamp)) {

            return createState(previousState.total + reading, previousState.count + 1, timestamp);
        }
        else
        {
            emitAverageEvent(measurementEvent, previousState);

            return createState(reading, 1, timestamp);
        }
    };

    return {
        init: init,
        update: update
    };
};

var measurementReadAveragPerDayCalculator = function measurementReadAveragPerDayCalculatorConstuctor($eventServices) {

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

var isFirstTime = function(previousState) {

return !previousState.lastTimestamp;

};

var isSameTimeSlot = function (currentTimestamp, lastTimestamp) {

return dailyTimestampFormatter.format(currentTimestamp)

== dailyTimestampFormatter.format(lastTimestamp);

};

var createEvent = function (timeStamp, reading, count) {

return {

Timeslot: dailyTimestampFormatter.format(timeStamp),

Total: reading,

Count: count,

Average: reading / count

};

var createState = function (reading, count, timestamp) {

return {

total: reading,

average: reading / count,

lastTimestamp: timestamp

};

var emitAverageEvent = function (measurementEvent, previousState) {

var streamName = 'MeasurementAverageDay-' + measurementEvent.streamId;

var event = createEvent(previousState.lastTimestamp, previousState.total, previousState.count);

eventServices.emit(streamName, "MeasurementAverageDay", event);

};

var init = function () {

return createState(0, 0, null);

};

var update = function (previousState, measurementEvent) {

var timestamp = measurementEvent.body.Timestamp;

var reading = measurementEvent.body.Reading;

if (isFirstTime(previousState) || isSameTimeSlot(timestamp, previousState.lastTimestamp)) {

return createState(previousState.total + reading, previousState.count + 1, timestamp);

}

else

{

emitAverageEvent(measurementEvent, previousState);

return createState(reading, 1, timestamp);

}

};

return {

init: init,

update: update

};

Just like previous projections, we instantiate the calculator module once globally and then reference it in the projection’s definition:

var calculator = measurementReadAveragPerDayCalculator();

fromCategory('Device')
    .foreachStream()
    .when({
        $init: calculator.init,
        MeasurementRead: calculator.update
    });

var calculator = measurementReadAveragPerDayCalculator();

fromCategory('Device')

.foreachStream()

.when({

$init: calculator.init,

MeasurementRead: calculator.update

});

Again we use the fromCategory(“Device”) definition to tell the engine that this projection’s events come from all streams in the Device category. Streams are categorized based on the stream name before the final dash by default: for example, Device-0 and Device-1 are both in the Device category. To enable categories, the $stream_by_category and $by_category system projections should be enabled. We delegate $init and MeasurementRead event handling to the calculator module, and we’re done.

Testing the projection

First, we test the formatting of the date part of the timestamp:

/// 
/// 

describe("when formatting the timeframe", function () {

    var formattedTimeSlot = function(value) {
        return dailyTimestampFormatter.format(value);
    };
    
    it("should contain year month and date", function () {

        expect(formattedTimeSlot("2000-01-01T08:02:39.687Z")).toMatch("2000/01/01");
        expect(formattedTimeSlot("2005-06-24T08:02:39.687Z")).toMatch("2005/06/24");
        expect(formattedTimeSlot("2012-10-10T08:02:39.687Z")).toMatch("2012/10/10");
        expect(formattedTimeSlot("2015-11-12T08:02:39.687Z")).toMatch("2015/11/12");
        expect(formattedTimeSlot("2030-12-18T08:02:39.687Z")).toMatch("2030/12/18");
    });
});

///

describe("when formatting the timeframe", function () {

var formattedTimeSlot = function(value) {

return dailyTimestampFormatter.format(value);

};

it("should contain year month and date", function () {

expect(formattedTimeSlot("2000-01-01T08:02:39.687Z")).toMatch("2000/01/01");

expect(formattedTimeSlot("2005-06-24T08:02:39.687Z")).toMatch("2005/06/24");

expect(formattedTimeSlot("2012-10-10T08:02:39.687Z")).toMatch("2012/10/10");

expect(formattedTimeSlot("2015-11-12T08:02:39.687Z")).toMatch("2015/11/12");

expect(formattedTimeSlot("2030-12-18T08:02:39.687Z")).toMatch("2030/12/18");

});

Next, we define the scenarios for testing the projection module:

describe("when projecting measurement reads average per day", function () {
    
    var handler;
    var defaultEvent;
    var projections;

    var eventEnvelope = function(timestamp, reading) {
        return { streamId: "device1", body: { Timestamp: timestamp, Reading: reading } };
    };

    beforeEach(function () {
        defaultEvent = { body: {} };
        projections = jasmine.createSpyObj('projections', ['emit']);
        handler = measurementReadAveragPerDayCalculator(projections);
    });

    describe("given state is empty", function () {
        
        it("should return initial state for first event", function() {

            var event = { body: { Timestamp: "2030-12-18T08:02:39.687Z", Reading: 1.23 } };
            var state = handler.init();
            
            var actcual = handler.update(state, event);

            expect(actcual).toEqual({
                total: 1.23,
                count: 1,
                average: 1.23,
                lastTimestamp: "2030-12-18T08:02:39.687Z"
            });
        });
    });

    describe("given events are in the same timeslot", function () {

        it("should return calculated average for all events", function() {

            var state = handler.init();
            state = handler.update(state, eventEnvelope("2030-12-18T08:00:39.687Z", 1.23));
            state = handler.update(state, eventEnvelope("2030-12-18T09:01:39.687Z", 2.34));
            state = handler.update(state, eventEnvelope("2030-12-18T10:02:39.687Z", 3.45));

            expect(state).toEqual({
                total: 7.02,
                count: 3,
                average: 2.34,
                lastTimestamp: "2030-12-18T10:02:39.687Z"
            });
        });
    });

    describe("given events are in the different timeslot", function () {

        var state;
        
        beforeEach(function () {
            state = handler.init();
            state = handler.update(state, eventEnvelope("2030-12-18T08:00:39.687Z", 1.23));
            state = handler.update(state, eventEnvelope("2030-12-18T09:01:39.687Z", 2.34));
            state = handler.update(state, eventEnvelope("2030-12-19T10:02:39.687Z", 3.44));
            state = handler.update(state, eventEnvelope("2030-12-19T11:02:39.687Z", 4.56));
        });

        it("an event with calculated average of first timeframe should be emitted", function () {
            expect(projections.emit).toHaveBeenCalledWith(
                'MeasurementAverageDay-device1',
                'MeasurementAverageDay',
                { Timeslot: '2030/12/18', Total: 3.57, Count: 2, Average: 1.785 }
            );
        });
        
        it("should return calculated average for events of new timeframe", function () {

            expect(state).toEqual({
                total: 8,
                count: 2,
                average: 4,
                lastTimestamp: "2030-12-19T11:02:39.687Z"
            });
        });
    });
});

describe("when projecting measurement reads average per day", function () {

var handler;

var defaultEvent;

var projections;

var eventEnvelope = function(timestamp, reading) {

return { streamId: "device1", body: { Timestamp: timestamp, Reading: reading } };

};

beforeEach(function () {

defaultEvent = { body: {} };

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

handler = measurementReadAveragPerDayCalculator(projections);

});

describe("given state is empty", function () {

it("should return initial state for first event", function() {

var event = { body: { Timestamp: "2030-12-18T08:02:39.687Z", Reading: 1.23 } };

var state = handler.init();

var actcual = handler.update(state, event);

expect(actcual).toEqual({

total: 1.23,

average: 1.23,

lastTimestamp: "2030-12-18T08:02:39.687Z"

});

describe("given events are in the same timeslot", function () {

it("should return calculated average for all events", function() {

var state = handler.init();

state = handler.update(state, eventEnvelope("2030-12-18T08:00:39.687Z", 1.23));

state = handler.update(state, eventEnvelope("2030-12-18T09:01:39.687Z", 2.34));

state = handler.update(state, eventEnvelope("2030-12-18T10:02:39.687Z", 3.45));

expect(state).toEqual({

total: 7.02,

average: 2.34,

lastTimestamp: "2030-12-18T10:02:39.687Z"

});

describe("given events are in the different timeslot", function () {

var state;

beforeEach(function () {

state = handler.init();

state = handler.update(state, eventEnvelope("2030-12-18T08:00:39.687Z", 1.23));

state = handler.update(state, eventEnvelope("2030-12-18T09:01:39.687Z", 2.34));

state = handler.update(state, eventEnvelope("2030-12-19T10:02:39.687Z", 3.44));

state = handler.update(state, eventEnvelope("2030-12-19T11:02:39.687Z", 4.56));

});

it("an event with calculated average of first timeframe should be emitted", function () {

expect(projections.emit).toHaveBeenCalledWith(

'MeasurementAverageDay-device1',

'MeasurementAverageDay',

{ Timeslot: '2030/12/18', Total: 3.57, Count: 2, Average: 1.785 }

);

});

it("should return calculated average for events of new timeframe", function () {

expect(state).toEqual({

total: 8,

average: 4,

lastTimestamp: "2030-12-19T11:02:39.687Z"

});

The first two examples are almost too simple to be worth writing tests for at all, but this example demonstrates how easy it is to test projections with JavaScript and Jasmine. All you have to do is call the projection module with the event stream as its input and then check whether the resulting state matches the expected state. We can also verify the emitted events by testing whether the mocked emit methods are called with the correct events.

Querying the results with the Client API

All new events are emitted to a stream, so reading them is straightforward:

var projectionResultStream = "MeasurementAverageDay-" + deviceStreamName;

var averages = _connection
    .ReadStreamEventsBackward(projectionResultStream);

var projectionResultStream = "MeasurementAverageDay-" + deviceStreamName;

var averages = _connection

.ReadStreamEventsBackward(projectionResultStream);

The implementation of the ReadStreamEventsBackward method is explained in the Client API post.

Conclusion

This a fairly naive implementation. First of all, the calculation doesn’t handle processing the same event twice properly: it will just add the value to the total twice, throwing off the average. It also expects events in some semblance of chronological order: if it receives an event whose timestamp indicates that a day has passed before it’s actually processed all the events for the day, then the emitted average event will be erroneous. Finally, it will fail to calculate the average entirely if a device stops sending messages during the day. To fix this, we’ll need to implement some kind of timeout mechanism. I’ll cover how to fix these issues in a future post.

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 – Partitioning events based on data found in previous events

2014-01-11/0 Comments/in Blog, EventStore /by Tim Cools

This example is a variation on the default indexing projection described on the Event Store blog. Another example of partition projections can be found on Rob Aston’s blog.

An indexing projection partitions events from multiple streams by linking said events to new streams to based on common properties. These new streams can be used as input for another projection or to retrieve the events from the store in a single read operation. In this example, all MeasurementRead events from all streams in the Device category are partitioned based on their DeviceType. The difference between this example and a basic indexing projection is that the DeviceType property used for the partitioning is not available in every event: it is only available in the first DeviceConfigured event that happens before the MeasurementRead events. So in order to create the index, we store the DeviceType in the projection’s state (per aggregate stream) when the projection handles the DeviceConfigured event. This allows the projection to link each subsequent MeasurementRead event to the corresponding DeviceType-* stream.

JavaScript implementation

The projection contains two methods: storeDeviceType, which stores the DeviceType in the state, and linkToDeviceType, which links the subsequent events to the DeviceType-* stream. This module accepts an $eventServices argument in its constructor which is used to pass in a mock object when the module is tested. This mock object is used to verify whether the linkTo method is called with the correct argument. When no $eventServices are passed in, the default Event Store linkTo implementation is used.

var deviceTypePartitioner = function ($eventServices) {

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

    var storeDeviceType = function (state, eventEnvelope) {
        state.deviceType = eventEnvelope.body.DeviceType;
        return state;
    };

    var linkToDeviceType = function(s, e) {

        if (!s.deviceType) return s;

        eventServices.linkTo("DeviceType-" + s.deviceType, e);

        return s;
    };
    
    return {
        storeDeviceType: storeDeviceType,
        linkToDeviceType: linkToDeviceType
    };
};

var deviceTypePartitioner = function ($eventServices) {

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

var storeDeviceType = function (state, eventEnvelope) {

state.deviceType = eventEnvelope.body.DeviceType;

return state;

};

var linkToDeviceType = function(s, e) {

if (!s.deviceType) return s;

eventServices.linkTo("DeviceType-" + s.deviceType, e);

return s;

};

return {

storeDeviceType: storeDeviceType,

linkToDeviceType: linkToDeviceType

};

The partitioner module is instantiated once globally and is referenced from the projection’s definition:

var partitioner = deviceTypePartitioner();

fromCategory("Device")
    .foreachStream()
    .when({

        DeviceConfigured: partitioner.storeDeviceType,
        MeasurementRead: partitioner.linkToDeviceType
    });

var partitioner = deviceTypePartitioner();

fromCategory("Device")

.foreachStream()

.when({

DeviceConfigured: partitioner.storeDeviceType,

MeasurementRead: partitioner.linkToDeviceType

});

The fromCategory(“Device”) definition tells the engine that this projection’s events come from all streams in the Device category. Streams are categorized based on the stream name before the final dash by default: for example, Device-0 and Device-1 are both in the Device category. To enable categories, the $stream_by_category and $by_category system projections should be enabled.

Next, we delegate the handling of the DeviceConfigured and MeasurementRead events to the partitioner module:

Testing the projection

/// 
/// 

describe("when partioning Device MeasurementRead events by type", function () {
    
    var partitioner;
    var projections;

    beforeEach(function () {
        projections = jasmine.createSpyObj('projections', ['linkTo']);
        partitioner = deviceTypePartitioner(projections);
    });

    describe("when storing devicetype", function () {

        it("should store the deviceType in the state", function () {
            var state = partitioner.storeDeviceType({}, { body: { DeviceType: 'someDeviceType' } });
            expect(state.deviceType).toEqual('someDeviceType');
        });
    });

    describe("when linking devicetype", function () {

        describe("given devicetype is not", function () {

            var state;

            beforeEach(function () {
                state = partitioner.linkToDeviceType({ }, { body: { } });
            });
            
            it("should return the previous state", function () {
                expect(state).toEqual({ });
            });
            
            it("should not link the event", function () {
                expect(projections.linkTo).wasNotCalled();
            });
        });

        describe("given devicetype is set", function () {

            var state;

            beforeEach(function () {
                state = partitioner.linkToDeviceType(
                    { deviceType: 'someDeviceType' },
                    { body: { someEvent: 'someDate' } });
            });

            it("should return the previous state", function () {
                expect(state).toEqual({ deviceType: 'someDeviceType' });
            });

            it("should link the event", function () {
                expect(projections.linkTo)
                    .wasCalledWith("DeviceType-someDeviceType", { body: { someEvent: 'someDate' } });
            });
        });
    });
});

///

describe("when partioning Device MeasurementRead events by type", function () {

var partitioner;

var projections;

beforeEach(function () {

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

partitioner = deviceTypePartitioner(projections);

});

describe("when storing devicetype", function () {

it("should store the deviceType in the state", function () {

var state = partitioner.storeDeviceType({}, { body: { DeviceType: 'someDeviceType' } });

expect(state.deviceType).toEqual('someDeviceType');

});

describe("when linking devicetype", function () {

describe("given devicetype is not", function () {

var state;

beforeEach(function () {

state = partitioner.linkToDeviceType({ }, { body: { } });

});

it("should return the previous state", function () {

expect(state).toEqual({ });

});

it("should not link the event", function () {

expect(projections.linkTo).wasNotCalled();

});

describe("given devicetype is set", function () {

var state;

beforeEach(function () {

state = partitioner.linkToDeviceType(

{ deviceType: 'someDeviceType' },

{ body: { someEvent: 'someDate' } });

});

it("should return the previous state", function () {

expect(state).toEqual({ deviceType: 'someDeviceType' });

});

it("should link the event", function () {

expect(projections.linkTo)

.wasCalledWith("DeviceType-someDeviceType", { body: { someEvent: 'someDate' } });

});

The first scenario tests how the DeviceType is stored and verifies that the state is updated correctly. The other scenarios test how the MeasurementRead events are linked, validating the cases where the DeviceType is not set and the resulting state after a call to the linkTo method of the $projectionServices mock object.

Even though this project requires practically no testing at all, it’s still a good example for how to use the Jasmine JavaScript test framework’s mock objects.

Querying the results with the Client API

Since the partitioned events are all linked to streams, reading them is straightforward. All we have to do is subscribe to the streams’ updates:

_connection.SubscribeToStream(
    "DeviceType-TV", 
    true, 
    (subscription, resolvedEvent) => // do something with the lineked events, 
    (subscription, subscriptionDropReason, exception) => // do something when subscription is dropped, 
    EventStoreCredentials.Default);

_connection.SubscribeToStream(

"DeviceType-TV",

true,

(subscription, resolvedEvent) => // do something with the lineked events,

(subscription, subscriptionDropReason, exception) => // do something when subscription is dropped,

EventStoreCredentials.Default);

In most cases, the output of an indexing projection will be used as the input for another projection. Combining projections in this fashion provides us with a powerful way to process events. Future examples will show you how this works in practice.

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:

EventStore Projections – Calculating daily averages

JavaScript implementation

Testing the projection

Querying the results with the Client API

Conclusion

Source code

Event Store Projections by Example

Event Store Projections – Partitioning events based on data found in previous events

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

EventStore Projections by Example

Base camp

Pages

Archive for month: January, 2014

JavaScript implementation

Testing the projection

Querying the results with the Client API

Conclusion

Source code

Event Store Projections by Example

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

EventStore Projections by Example

Base camp

Blog – Tags

Pages