Archive for month: December, 2013

Event Store Projections – Counting events of a specific type

2013-12-30/0 Comments/in Blog, EventStore /by Tim Cools

Our first Event Store Projections example simply counts the number of events of a specific type. Pretty mundane, I know, but it’s a good place to start.

In this example, the projection will take all MeasurementRead events from all streams and increase a counter in its global state for each one. The end result is a global state containing the number of MeasurementRead events.

JavaScript implementation

The projection logic is implemented in a self-revealing module. This pattern allows us to decide which methods of a module should be accessible from outside, much like creating a class with public and private methods. The pattern is probably overkill for this projection, since it only has public methods, but it allows us to create unit tests for its logic:

var eventCounter = function () {

    var init = function () {
        return { count: 0 };
    };

    var increase = function (state, eventEnvelope) {

        state.count += 1;

        return state;
    };

    return {
        init: init,
        increase: increase
    };
};

var eventCounter = function () {

var init = function () {

return { count: 0 };

};

var increase = function (state, eventEnvelope) {

state.count += 1;

return state;

};

return {

init: init,

increase: increase

};

The init method initializes the state for the first time by setting the count property to 0. The increase method increases the count property by 1 when called.

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

var counter = eventCounter();

options({
    producesResults: true
});

fromAll()
    .when({
        $init: counter.init,
        MeasurementRead: counter.increase
    });

var counter = eventCounter();

options({

producesResults: true

});

fromAll()

.when({

$init: counter.init,

MeasurementRead: counter.increase

});

The fromAll definition tells the engine that the projection’s events come from all streams. Inside, we delegate $init and MeasurementRead to our counter module implementation. The $init field defines the method that initializes the projection’s state, and the MeasurementRead field defines that all MeasurementRead events are to be handled by the counter.increase method.

producesResults is set to true so that the projection’s state is stored in a stream. This allows us to subscribe to projection state updates without polling and use the output of the projection as input for other projections, as I’ll demonstrate later. In the future, we can replace this with the outputState method.

Testing the projection

I used Jasmine to test this logic. Jasmine is a really nice BDD framework for testing JavaScript code that encourages developers to write readable test scenarios:

/// 
/// 

describe("when counting measurement reads", function () {

    var defaultEvent = { body: {} };
    var counter;

    beforeEach(function () {
        counter = eventCounter();
    });

    it("should initialize new state with count 0", function () {

        var initState = counter.init();

        expect(initState.count).toEqual(0);
    });

    it("should count the number of events", function () {

        var state = counter.init();
        state = counter.increase(state, defaultEvent);
        state = counter.increase(state, defaultEvent);
        state = counter.increase(state, defaultEvent);

        expect(state.count).toEqual(3);
    });
});

///

describe("when counting measurement reads", function () {

var defaultEvent = { body: {} };

var counter;

beforeEach(function () {

counter = eventCounter();

});

it("should initialize new state with count 0", function () {

var initState = counter.init();

expect(initState.count).toEqual(0);

});

it("should count the number of events", function () {

var state = counter.init();

state = counter.increase(state, defaultEvent);

expect(state.count).toEqual(3);

});

The test contains two cases: the first one verifies the initial state of the projection, and the second one verifies the count property is increased by 1 for each call of the increase method. The first two lines define the referenced JavaScript files, which enables Intellisense in Visual Studio and allows R# to run the tests from your IDE.

Querying the results with the Client API

To read the value of the projection, we use the getState method of the ProjectionsManager class.

Since we specified that the output of the state should be stored in a stream, we can subscribe to event updates without polling:

_connection.SubscribeToStream(
    "$projections-MeasurementReadCounter-result", 
    false, 
    (subscription, resolvedEvent) => // do something with the state updates, 
    (subscription, subscriptionDropReason, exception) => // do something when subscription is dropped, 
    EventStoreCredentials.Default);

_connection.SubscribeToStream(

"$projections-MeasurementReadCounter-result",

false,

(subscription, resolvedEvent) => // do something with the state updates,

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

EventStoreCredentials.Default);

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 Client API Basics (C#)

2013-12-22/2 Comments/in Blog, EventStore /by Tim Cools

Here I’ll give a overview on how to use the Event Store Client API from C#. I’ll be highlighting the features used to implement the examples described in the following blog posts. If you’re only interested in projections written in JavaScript, you can safely skip this post and go directly to the first example.

Running Event Store

These examples are based on the latest main branch version available at the time of writing (RC3.0). Since you have to run the Client API against the same version of Event Store, I’ve included the binaries in the github repository. This means you can check out the examples and run them without worrying about version compatibility or building Event Store yourself.

To run the examples, run Event Store as an administrator with –run-projections=all. The following command will run Event Store with projections enabled and an in-memory database:

lib\EventStore\EventStore.SingleNode.exe --run-projections=all --mem-db

1	lib\EventStore\EventStore.SingleNode.exe --run-projections=all --mem-db

Event Store Client API Connections

Event Store is accessible through two protocols: HTTP and TCP. The default HTTP endpoint is port 1113 and the default TCP endpoint to 2113, though both of these can be modified from the command line using –tcp-port and –http-port:

EventStore.SingleNode.exe --tcp-port=5555 --http-port=5556

1	EventStore.SingleNode.exe --tcp-port=5555 --http-port=5556

Creating a connection

The Event Store Client API connection is accessible through the IEventStoreConnection interface and connects to the TCP endpoint:

internal class IPEndPointFactory
{
    public static IPEndPoint DefaultTcp()
    {
        return CreateIPEndPoint(1113);
    }

    private static IPEndPoint CreateIPEndPoint(int port)
    {
        var address = IPAddress.Parse("127.0.0.1");
        return new IPEndPoint(address, port);
    }
}

public static class EventStoreConnectionFactory
{
    public static IEventStoreConnection Default()
    {
        var connection = EventStoreConnection.Create(IPEndPointFactory.DefaultTcp());
        connection.Connect();
        return connection;
    }
}

internal class IPEndPointFactory

{

public static IPEndPoint DefaultTcp()

{

return CreateIPEndPoint(1113);

}

private static IPEndPoint CreateIPEndPoint(int port)

{

var address = IPAddress.Parse("127.0.0.1");

return new IPEndPoint(address, port);

}

public static class EventStoreConnectionFactory

{

public static IEventStoreConnection Default()

{

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

connection.Connect();

return connection;

}

Most connection methods accept credentials for connecting to Event Store. These examples always use the default admin account:

public class EventStoreCredentials
{
    private static readonly UserCredentials _credentials = 
        new UserCredentials("admin", "changeit");

    public static UserCredentials Default { get { return _credentials; } }
}

public class EventStoreCredentials

{

private static readonly UserCredentials _credentials =

new UserCredentials("admin", "changeit");

public static UserCredentials Default { get { return _credentials; } }

}

Converting events to and from JSON

These examples use static-typed events which are serialized to JSON when they are sent to Event Store. When they are retrieved from the database, they are deserialized from JSON into static-typed events. Two extension methods take care of this process:

public static EventData AsJson(this object value)
{
    if (value == null) throw new ArgumentNullException("value");

    var json = JsonConvert.SerializeObject(value);
    var data = Encoding.UTF8.GetBytes(json);
    var eventName = value.GetType().Name;

    return new EventData(Guid.NewGuid(), eventName, true, data, new byte[] { });
}

public static T ParseJson<T>(this RecordedEvent data)
{
    if (data == null) throw new ArgumentNullException("data");

    var value = Encoding.UTF8.GetString(data.Data);

    return JsonConvert.DeserializeObject<T>(value);
}

public static EventData AsJson(this object value)

{

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

var json = JsonConvert.SerializeObject(value);

var data = Encoding.UTF8.GetBytes(json);

var eventName = value.GetType().Name;

return new EventData(Guid.NewGuid(), eventName, true, data, new byte[] { });

}

public static T ParseJson<T>(this RecordedEvent data)

{

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

var value = Encoding.UTF8.GetString(data.Data);

return JsonConvert.DeserializeObject<T>(value);

}

These methods use JSON.NET for JSON serialization.

Appending an event to a stream

Now that we’ve created a connection and know how to serialize and deserialize events, we can append an event to a stream. We don’t have to create the stream first in this case: if we try to append an event to a stream that doesn’t exist, the stream is automatically created for us.

var @event = new MeasurementRead(DateTime.Now, 10.12m)
    .AsJson();

_connection.AppendToStream(deviceName, ExpectedVersion.Any, new[] { @event });

var @event = new MeasurementRead(DateTime.Now, 10.12m)

.AsJson();

_connection.AppendToStream(deviceName, ExpectedVersion.Any, new[] { @event });

Note that concurrency control in Event Store is handled through versioning. In this example, we skip concurrency control by passing ExpectedVersion.Any when an event is appended to a stream.

Getting the last event number from a stream

To read events from a stream in reverse order, we need to know the number of the last event in the stream. To get this number, we ask for the last event of the stream by passing an EventNumber of -1 when calling the ReadEvent method:

private static int? GetLastEventNumber(this IEventStoreConnection connection, string streamName)
{
    var lastEvent = connection.ReadEvent(streamName, -1, false, EventStoreCredentials.Default);
    if (lastEvent == null || lastEvent.Event == null) return null;

    return lastEvent.Event.Value.OriginalEventNumber;
}

private static int? GetLastEventNumber(this IEventStoreConnection connection, string streamName)

{

var lastEvent = connection.ReadEvent(streamName, -1, false, EventStoreCredentials.Default);

if (lastEvent == null || lastEvent.Event == null) return null;

return lastEvent.Event.Value.OriginalEventNumber;

}

Getting events from a stream in reverse order

Now we can read events in reverse order starting from the number of the last event. This example reads events in pages of 10 at a time:

private const int PageSize = 10;

public static IEnumerable<T> ReadStreamEventsBackward<T>(this IEventStoreConnection connection, string streamName)
{
    if (connection == null) throw new ArgumentNullException("connection");

    var lastEventNumber = connection.GetLastEventNumber(streamName);

    return lastEventNumber == null 
        ? new T[0] 
        : ReadResult<T>(connection, streamName, lastEventNumber.Value);
}

private static IEnumerable<T> ReadResult<T>(IEventStoreConnection connection, string streamName, int lastEventNumber)
{
    var result = new List<T>();

    do
    {
        var events = connection.ReadStreamEventsBackward(streamName, lastEventNumber, PageSize, false, EventStoreCredentials.Default);

        result.AddRange(events.Events.Select(e => e.ParseJson<T>()));
        lastEventNumber = events.NextEventNumber;

    } while (lastEventNumber != -1);

    return result;
}

private const int PageSize = 10;

public static IEnumerable<T> ReadStreamEventsBackward<T>(this IEventStoreConnection connection, string streamName)

{

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

var lastEventNumber = connection.GetLastEventNumber(streamName);

return lastEventNumber == null

? new T[0]

: ReadResult<T>(connection, streamName, lastEventNumber.Value);

}

private static IEnumerable<T> ReadResult<T>(IEventStoreConnection connection, string streamName, int lastEventNumber)

{

var result = new List<T>();

{

var events = connection.ReadStreamEventsBackward(streamName, lastEventNumber, PageSize, false, EventStoreCredentials.Default);

result.AddRange(events.Events.Select(e => e.ParseJson<T>()));

lastEventNumber = events.NextEventNumber;

} while (lastEventNumber != -1);

return result;

}

Subscribing to all new events

We can use the connection’s SubscribeToAll method to subscribe to all new events. This method is called each time a new event is stored in Event Store:

public void StartReading()
{
    _connection.SubscribeToAll(true, Appeared, Dropped, EventStoreCredentials.Default);
}

private void Appeared(EventStoreSubscription subscription, ResolvedEvent resolvedEvent)
{
    // do something with the events here
    // var @event = resolvedEvent.ParseJson();
}

private void Dropped(EventStoreSubscription subscription, SubscriptionDropReason subscriptionDropReason, Exception exception)
{
    // is called when the tcp connection is dropped, we could
    // implement recovery here
}

public void StartReading()

{

_connection.SubscribeToAll(true, Appeared, Dropped, EventStoreCredentials.Default);

}

private void Appeared(EventStoreSubscription subscription, ResolvedEvent resolvedEvent)

{

// do something with the events here

// var @event = resolvedEvent.ParseJson();

}

private void Dropped(EventStoreSubscription subscription, SubscriptionDropReason subscriptionDropReason, Exception exception)

{

// is called when the tcp connection is dropped, we could

// implement recovery here

}

Subscribing to new events from a specific stream

Similarly to SubscribeToAll, we can also use SubscribeToStream to retrieve events for a specific stream:

public void SubscribeValueChange(string eventName)
{
    _connection.SubscribeToStream(eventName, false, ValueChanged, Dropped, EventStoreCredentials.Default);
}

private void ValueChanged(EventStoreSubscription eventStoreSubscription, ResolvedEvent resolvedEvent)
{
    // do something with the events here
    // var @event = resolvedEvent.ParseJson();
}

private void Dropped(EventStoreSubscription subscription, SubscriptionDropReason subscriptionDropReason, Exception exception)
{
    // is called when the tcp connection is dropped, we could
    // implement recovery here
}

public void SubscribeValueChange(string eventName)

{

_connection.SubscribeToStream(eventName, false, ValueChanged, Dropped, EventStoreCredentials.Default);

}

private void ValueChanged(EventStoreSubscription eventStoreSubscription, ResolvedEvent resolvedEvent)

{

// do something with the events here

// var @event = resolvedEvent.ParseJson();

}

private void Dropped(EventStoreSubscription subscription, SubscriptionDropReason subscriptionDropReason, Exception exception)

{

// is called when the tcp connection is dropped, we could

// implement recovery here

}

Managing projections

We can manage projections using the ProjectionsManager class of the Client API. This class allows us to retrieve information about projections as well as enable, create, and update them.

Creating a ProjectionsManager

This ProjectionsManager will use HTTP (default port 2113) to connect to Event Store:

internal class IPEndPointFactory
{
    public static IPEndPoint DefaultHttp()
    {
        return CreateIPEndPoint(2113);
    }

    private static IPEndPoint CreateIPEndPoint(int port)
    {
        var address = IPAddress.Parse("127.0.0.1");
        return new IPEndPoint(address, port);
    }
}

var logger = new ConsoleLogger();
var projectionsManager = new ProjectionsManager(logger, IPEndPointFactory.DefaultHttp());

internal class IPEndPointFactory

{

public static IPEndPoint DefaultHttp()

{

return CreateIPEndPoint(2113);

}

private static IPEndPoint CreateIPEndPoint(int port)

{

var address = IPAddress.Parse("127.0.0.1");

return new IPEndPoint(address, port);

}

var logger = new ConsoleLogger();

var projectionsManager = new ProjectionsManager(logger, IPEndPointFactory.DefaultHttp());

In the example, we’re using the ConsoleLogger provided by the Client API. You can replace it with your own logger if necessary.

Manipulating projections

The methods of the ProjectionsManager class are fairly straightforward:

var all = projectionsManager.ListAll(EventStoreCredentials.Default);
var json = JsonConvert.DeserializeObject(all);

foreach (var projection in json.projections)
{
    var name = projection.name.Value;
    var status = projection.status.Value;

    //Do something with the projection
}

var all = projectionsManager.ListAll(EventStoreCredentials.Default);

var json = JsonConvert.DeserializeObject(all);

foreach (var projection in json.projections)

{

var name = projection.name.Value;

var status = projection.status.Value;

//Do something with the projection

}

var name = "MyNewContinuousProjection";
var projection = @"fromAll().when({ $any: function (s, e) { 
                   s.lastEventName = e.eventType; return s; } } )";

projectionsManager.CreateContinuous(name, projection, EventStoreCredentials.Default);

var name = "MyNewContinuousProjection";

var projection = @"fromAll().when({ $any: function (s, e) {

s.lastEventName = e.eventType; return s; } } )";

projectionsManager.CreateContinuous(name, projection, EventStoreCredentials.Default);

var name = "MyNewContinuousProjection";
var expectedQuery = @"fromAll().when({ $any: function (s, e) { 
                      s.lastEventName = e.eventType; return s; } } )";

var currentQuery = _projectionsManager.GetQuery(name, EventStoreCredentials.Default);

if (expectedQuery != currentQuery)
{
    _projectionsManager.UpdateQuery(name, expectedQuery, EventStoreCredentials.Default);
}

var name = "MyNewContinuousProjection";

var expectedQuery = @"fromAll().when({ $any: function (s, e) {

s.lastEventName = e.eventType; return s; } } )";

var currentQuery = _projectionsManager.GetQuery(name, EventStoreCredentials.Default);

if (expectedQuery != currentQuery)

{

_projectionsManager.UpdateQuery(name, expectedQuery, EventStoreCredentials.Default);

}

public T GetState(string projectionName)
{
    var state = _projectionsManager.GetState(projectionName);
    return state.ParseJson();
}

public T GetState(string projectionName)

{

var state = _projectionsManager.GetState(projectionName);

return state.ParseJson();

}

Subscribing to a projection’s state change

We can store the state of an event in a stream. This allows us to subscribe to the stream and retrieve updates when its state changes. If there’s only one state stored for all projected streams, the stream with the state is named “$projections-ProjectionName-result”, where ProjectionName is the name of the stream.

_connection.SubscribeToStream($projections-MeasurementReadCounter-result", 
    false, 
    ValueChanged, 
    Dropped, 
    EventStoreCredentials.Default);

_connection.SubscribeToStream($projections-MeasurementReadCounter-result",

false,

ValueChanged,

Dropped,

EventStoreCredentials.Default);

Keep in mind that the projection needs to know that it should store the state in the stream. We can do this by setting the producesResults option to true or by calling the outputState method in the JavaScript projection definition:

options({ producesResults: true });

// or 

fromAll() 
    .when({
        // ...
    })
    .outputState()

options({ producesResults: true });

// or

fromAll()

.when({

// ...

})

.outputState()

Source code

A working project for these examples 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 By Example

2013-12-21/0 Comments/in Blog, EventStore /by Tim Cools

I’ve spent some time recently exploring Event Store Projections written in JavaScript. Since there isn’t a lot of documentation yet, I’ll be publishing my findings in a series of blog posts.

Event Store Projections

Event Store is an open-source functional database that stores data as a series of immutable events. The concept isn’t new: logs have been used to store data in this exact fashion for decades, but it’s still garnered much attention from the DDD community in the past few years, especially from people implementing CQRS+ES. Event Store Projections allow us to process events, filter events, and even produce new events, features we can leverage for event processing in systems based on Event-Driven Architecture.

Why JavaScript?

Although it’s possible to write projections in any language, Event Store supports JavaScript natively. This means we don’t have to rely on external programs to process our projections if we write them in JavaScript, and we don’t have to manage their progress and state manually.

I used to hate having to deal with JavaScript; in most projects where I’ve seen it used, it’s an absolute mess. But the language has matured in the past few years: now we have test frameworks, design patterns, and guidelines to work with. Thanks to these tools, I’ve actually learned to like programming in JavaScript.

Examples

In the following posts, I provide examples of solutions to real-world problems I’ve faced using JavaScript and Event Store Projections. These include developing maintainable and testable code, filtering data, and calculating averages. Here are the topics I intend to write about in the future:

Source code

The source for these examples can be found on github:

https://github.com/tim-cools/EventStore-Examples

Be sure to read the README file first.

Event Store Projections – Counting events of a specific type

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

Event Store Projections by Example

Event Store Client API Basics (C#)

Running Event Store

Event Store Client API Connections

Creating a connection

Converting events to and from JSON

Appending an event to a stream

Getting the last event number from a stream

Getting events from a stream in reverse order

Subscribing to all new events

Subscribing to new events from a specific stream

Managing projections

Creating a ProjectionsManager

Manipulating projections

Subscribing to a projection’s state change

Source code

Event Store Projections by Example

Event Store Projections By Example

Event Store Projections

Why JavaScript?

Examples

Source code

More Event Store Projections documentation

Base camp

Pages

Archive for month: December, 2013

JavaScript implementation

Testing the projection

Querying the results with the Client API

Source code

Event Store Projections by Example

Running Event Store

Event Store Client API Connections

Creating a connection

Converting events to and from JSON

Appending an event to a stream

Getting the last event number from a stream

Getting events from a stream in reverse order

Subscribing to all new events

Subscribing to new events from a specific stream

Managing projections

Creating a ProjectionsManager

Manipulating projections

Subscribing to a projection’s state change

Source code

Event Store Projections by Example

Event Store Projections

Why JavaScript?

Examples

Source code

More Event Store Projections documentation

Base camp

Blog – Tags

Pages