erles - Erlang client for Event Store

erles is an Erlang client library for
Event Store.

erles is licensed under The MIT License.
See LICENSE file for details.

Copyright © 2013 Andrii Nakryiko

I'm very new to Erlang ecosystem and this is my first Erlang project, so if I'm
doing something in a non-optimal, non-idiomatic or, even worse, plain wrong way,
please, let me know and I'll try to fix it.

Supported versions of Event Store

As erles uses and supports some of nice latest changes to Event Store (next
expected versions on writes and soft deletes), which haven't been released yet,
erles can be used only with latest dev-builds currently. As soon as Event Store
releases its new version, erles will stick to supporting stable versions.

Erles itself is not yet production ready, so there could be some slight breaking
changes if that will lead to better client library.


erles aims to provide a convenient and reliable client for working with Event
Store. It strives to be on par or better (and more convenient) with official
Event Store client functionality-wise.

Main features:

  • Permanent connection to node/cluster with automatic reconnects.
  • User authentication support;
  • Automatic reconnection to master node for maximum performance and least
    network overhead.
  • Batch writes to stream.
  • Explicit transaction writes.
  • Stream deletion (both soft- and hard-deletes).
  • Single event reads from streams.
  • Stream reads in forward/backward direction from usual stream or $all
  • Permanent catch-up subscriptions which catch-up, subscribe to live events,
    and reconnects on errors/connection drops.
  • Primitive subscriptions provided by Event Store directly for those who need
    full control.
  • Metadata set/get operations which allow to work with both raw and structured
    stream metadata.
  • Ping operation (just for fun and testing that Event Store is responding :).

Using erles


Download sources from Bitbucket repository.

To build library and run tests:

$ rebar get-deps
$ rebar compile
$ rebar skip_deps=true eunit

You can also add repository as a dependency to rebar.config:

{deps, [
    {erles, ".*", {hg, "", {branch, "master"}}}

erles requires some of standard application to be started, so if you are working
with it in Erlang shell, please first run erles:start(). There is
also a convenience shell.cmd/ scripts that start Erlang console with
correct ebin paths set.

If you are doing proper Erlang release, adding erles to applications
property of your .app file should just work.

General conventions


Whenever string value is expected, erles accepts both <<"binary strings">> and
"list strings". If not - that's a bug, please report it.

Operation results

All operations in case of success return either ok atom or a tuple in which
first element is atom ok. Also most operations in case of failure return a
2-tuple {error, Reason}, where Reason is either predefined atom or any
Erlang term (only in some rare cases).

Common operation failure reasons:

  • retry_limit - operation reached allowed number of retries;
  • server_timeout - no response from server was returned within allotted
  • {aborted, AbortReason} - operation was aborted due to closed erles
    connection (because of AbortReason).
  • {not_authenticated, EsMsg} - provided login/password was not authenticated
    on server. EsMsg can provide a bit more information;
  • {bad_request, EsMsg} - Event Store received malformed package for some
    reason (e.g., because of protocol version mismatch). EsMsg can provide
    more information.

For more detail information on structure and type of arguments and results,
please refer to erles.hrl header file (which you probably will want to
include into your modules for more convenient usage of erles records).

Operation options

Most erles functions exist in at least two forms:

  • short one with sensible defaults;

  • the long one, which takes a list of additional options that modify or
    enhance operation behavior.

Connection operation accepts the following options:

  • {default_auth, noauth | defauth | {Login, Pass}} (default: noauth) -
    default authentication used for each operation if no explicit override is
    provided for particular operation:

    • noauth - no authentication should be used;

    • defauth - same as noauth (when set as connection option, see
      below for meaning when used for individual operation);

    • {Login, Pass} - login/password string pair to use for authentication.

  • {max_server_ops, pos_integer()} (default: 2000) - maximum amount of
    active operations on Event Store node. Used to prevent node overload. All
    other operations are enqueued internally in erles connection.

  • {operation_timeout, pos_integer()} (default: 7000) - timeout after which
    {error, server_timeout} will be returned for operation, if there was no
    response from Event Store.

  • {operation_retries, non_neg_integer()} (default: 10) - how many times
    operation will be retried (due to reconnection, prepare/commit/forwarding
    timeout, etc) before giving up and returning {error, retry_limit}.

  • {retry_delay, non_neg_integer()} (default: 500) - the delay before
    operation retry.

  • {connection_timeout, pos_integer()} (default: 2000) - TCP connection

  • {reconnection_delay, non_neg_integer()} (default: 500) - the delay
    before TCP reconnection in case of dropped connection.

  • {max_conn_retries, non_neg_integer()} (default: 10) - how many times
    erles will try to reconnect to a given destination (node or cluster).

  • {heartbeat_period, pos_integer()} (default: 500) - heartbeat period.

  • {heartbeat_timeout, pos_integer()} (default: 1000) - heartbeat timeout
    after which connection is declared closed, if no heartbeat response is get
    from server.

  • {dns_timeout, pos_integer()} (default: 2000) - timeout of resolving DNS

  • {gossip_timeout, pos_integer()}. (default: 1000) - timeout of getting
    gossip from node.

All timeouts are in milliseconds. infinity is not accepted.

Write operations (append, transactions, stream deletion, setting stream
metadata) accept the following options:

  • {auth, noauth | defauth | {Login, Pass}} (default: defauth) -
    authentication used for operation.

    • noauth - no authentication should be used;

    • defauth - use default authentication of connection (set with
      default_auth option provided in connect/4). If no default
      authentication is set on connection level, noauth is used;

    • {Login, Pass} - login/password string pair to use for authentication.

  • {master_only, boolean()} - the flag that specifies whether operation
    should be performed exclusively on master node. When master_only flag is
    set, erles connection will automatically reconnect to new master in case
    connected node is not master. This option ensures no additional
    communication overhead and lower latency for writes (otherwise write
    operations are forwarded within Event Store nodes internally) and most up
    to date reads from master for reads (otherwise reads are served locally
    from Event Store node, no forwarding occurs). Default value is true.

Read operations (single event read, stream range reads, getting stream
metadata) accept the following options:

  • auth - same as for write operations;

  • master_only - same as for write operations, see option's description
    for effect on reads;

  • {resolve, boolean()} - the flag that indicates whether Event Store
    should resolve links emitted from projections into actual events. If the
    flag is set - any link is automatically replaced with original linked
    event. Defaults to true.

Subscription operation accepts the following options:

  • auth - same as for write and read operations;

  • resolve - same as for read operations;

  • {subscriber, pid()} - PID of process that will receive all subscription
    messages. If not provided - the caller process ID is used.

  • {read_batch, pos_integer()} - the number of events that will be read
    in a single stream read operation during catch-up phase of subscription.
    Default value - 100.


All operations in erles are performed through a connection, which, once
established, stays open and keeps reconnecting in case of TCP connection drop.

erles connection is established through calling connect/2 (or connect/3)
function. When successful, function returns process ID, which is used with all
operations (except subscription unsubscribe/1).

-spec connect('node', {Ip :: inet:ip_address(), TcpPort :: 0..65535})              -> {'ok', pid()} | {'error', term()};
             ('dns', {ClusterDns :: string() | binary(), ManagerPort :: 0..65535}) -> {'ok', pid()} | {'error', term()};
             ('cluster', [{Ip :: inet:ip_address(), HttpPort :: 0..65535}])        -> {'ok', pid()} | {'error', term()}.

Three connection modes are currently supported:

  • Connection to single node. IP address and TCP port of Event Store node
    should be specified:
> {ok, C} = erles:connect(node, {{127,0,0,1}, 1113}).
  • Connection to cluster of nodes. A list of 2-tuples with IP address and
    HTTP port of each node is specified. erles will automatically determine
    through gossip protocol the best node to connect to.
> {ok, C} = erles:connect(cluster, [{{127,0,0,1}, 7777},
                                    {{127,0,0,1}, 8888},
                                    {{127,0,0,1}, 9999}]).
  • Connection to cluster of nodes through specifying DNS name and common HTTP
    port for all endpoints. erles will resolve DNS name into a list of IP
    addresses and pair them with provided HTTP port forming same list of
    endpoints used with cluster connection mode. This mode is intended to be
    used with Event Store's managers, though specifying HTTP port of cluster
    node is enough to function properly provided all machines have nodes with
    same HTTP port. Usage:
> {ok, C} = erles:connect(dns, {<<"">>, 30778}).

To close existing connection just call close/1:

> erles:close(C).

Writing events

Event Store provides two ways to store events, both of which are supported by

  • batch writes when all sent events are committed immediately;

  • transactional writes. You have to start transaction specifying expected
    version of stream you write to, then you can make multiple transactional
    writes of events which Event Store won't commit until you explicitly
    commit that transaction.

Batch writes

Batch writes are done with append/4 (append/5 for variant with options)

-spec append(pid(), stream_id(), exp_ver(), [event_data()]) ->
        {'ok', stream_ver()} | {'error', write_error()}.

Where parameters (in order) are:

  1. erles connection PID (returned from connect/2 function).
  2. Non-empty stream ID (either binary or list string).
  3. Expected version of stream we are about to write to. Expected version is
    either atom any (to specify that you don't care about version of
    stream, so no optimistic concurrency control should be used and write
    should always succeed) or integer, with -2 meaning same as any, -1
    meaning empty stream, any other non-negative integer specifying the
    number of last event in the stream.
  4. List of events to append to stream, each event defined as:
            event_id = erles_utils:gen_uuid()                 :: uuid(),
            event_type = erlang:error({required, event_type}) :: binary() | string(),
            data_type = raw                                   :: 'raw' | 'json',
            data = erlang:error({required, data})             :: binary(),
            metadata = <<>>                                   :: binary()

The two required fields are event_type and data. event_id, if not
specified, will be set to newly generated UUID, metadata will be empty
and data type will be assumed to be raw (not structured). If you are
writing valid JSON as data, you should set data_type to json, so
Event Store's projections will know how to work with event data.

In case of success, the new expected version of stream is returned (that should
be used with consequent writes to same stream).

The write-specific failure reasons include:

  • wrong_exp_ver - provided stream expected version is wrong;

  • access_denied - you don't have enough access rights for operation, provide
    login/password credentials with enough access rights;

  • stream_deleted - you are trying to write events into the stream that was
    permanently deleted;

  • invalid_transaction - this can mean few things, but most probably wrong
    transaction ID (for transaction operations).


> {ok, NextExpVer} = erles:append(C, <<"stream">>, any,
                                  [#event_data{event_type="type1", data="data1"},
                                   #event_data{event_type="type2", data="data2"}]).

Transactional writes

To start transaction you use txn_start/3 (txn_start/4) function:

-spec txn_start(pid(), stream_id(), exp_ver()) ->
        {'ok', trans_id()} | {'error', write_error()}.

All parameters have the same meaning as with append/4. In case of success
transaction ID is returned which should be used with all operations withing that
transaction. Failure reasons are same as for append/4.


> {ok, Tid} = erles:txn_start(C, <<"stream">>, 1).

To write some events within transaction you use txn_append/3 function:

-spec txn_append(pid(), trans_id(), [event_data()]) ->
        'ok' | {'error', write_error()}.

It accepts connection PID, transaction ID returned from txn_start/3 and a list
of event of the same structure as in append/4. When operation succeeds, events
are not committed yet.


> ok = erles:txn_append(C, Tid, [#event_data{event_type = <<"et1">>,
                                             data = <<"{\"num\":123}">>,
                                             data_type = json}]).
> ok = erles:txn_append(C, Tid, [#event_data{event_type = <<"et2">>,
                                             data = <<1,2,3>>}]).

To force events to be stored in the stream, call txn_commit/2:

-spec txn_commit(pid(), trans_id()) ->
        {'ok', stream_ver()} | {'error', write_error()}.

The result is the same as with append/4 - expected version of stream for
following writes. Failures are same as well.


> {ok, NextExpVer} = erles:txn_commit(C, Tid).

Reading events

To be written... For now, please take a look at erles_tests.erl for some


To be written... For now, please take a look at erles_tests.erl for some

Working with stream metadata

To be written... For now, please take a look at erles_tests.erl for some