eventful.core — eventful 0.1.0-alpha6

Event Store (https://eventstore.org/) client library. Wrapper around the JVM Client (TCP interface).

Provides (hopefully) Clojure-friendly functions to work with the Event Store. Supports Transit-JSON, EDN & JSON formats as well as raw bytes. It's easy to add more.

This namespace covers (almost) all of the API of the JVM Client and this documentation often copies its documentation when it may be helpful.

close-subscription clj

(close-subscription sub)

Closes a subscription returned by the subscribe-to-stream, subscribe-to-all-streams or persistently-subscribe fns. Returns nil.

connect clj

(connect {:keys [hostname port login password conn-timeout-ms max-reconn
                 reconn-del-min-ms reconn-del-max-ms heartb-int-ms
                 heartb-timeout-ms max-retries timeout-ms read-batch-sz buf-sz
                 overflow-strat cluster-settings serial-parallel serial-ordered
                 conn-name]})

Returns a full duplex connection to the event store.

All operations are handled in a full async manner. Many threads can use this connection at the same time or a single thread can make many asynchronous requests. To get the most performance out of the connection it is generally recommended to use it in this way.

Options: :hostname - (required) the event store address :port - ” :login - (required) user credentials to perform operation with :password - ” :conn-timeout-ms - the desired connection timeout (milliseconds, 1000 by default) :max-reconn - maximum number of reconnections before backing off, -1 to reconnect forever (100 by default) :reconn-del-min-ms - delay before first reconnection (milliseconds, 250 by default) :reconn-del-max-ms - maximum delay on reconnections (milliseconds, 10000 by default) :heartb-int-ms - the interval at which to send heartbeat messages (milliseconds, 500 by default) :heartb-timeout-ms - the interval after which an unacknowledged heartbeat will cause the connection to be considered faulted and disconnect (milliseconds, 5000 by default) :max-retries - the maximum number of operation retries (10 by default) :timeout-ms - the amount of time before an operation is considered to have timed out (milliseconds, 30000 by default) :read-batch-sz - number of events to be retrieved by client as single message (500 by default) :buf-sz - the size of the buffer in element count (100000 by default) :overflow-strat - strategy that is used when elements cannot fit inside the buffer (:drop-head, :drop-tail, :drop-buf, :drop-new or :fail which is the default) :cluster-settings - contains settings relating to a connection to a cluster (can be created with eventstore.j.ClusterSettingsBuilder) :serial-parallel - the number of serialization/deserialization functions to be run in parallel (8 by default) :serial-ordered - serialization done asynchronously and these futures may complete in any order, but results will be used with preserved order if set to true (the default) :conn-name - client identifier used to show a friendly name of client in the event store ("jvm-client" by default)

create-persistent-subscription clj

(create-persistent-subscription {:keys [conn stream group] :as m} settings)

Asynchronously creates a persistent subscription group on a stream.

Options: :conn - (required) see write-events fn :stream - ” :group - (required) the name of a group to create, e.g. "foo" :login - see write-events fn :password - ”

settings is a map with optional values for: :resolve-link-tos - see read-event fn :from - see subscribe-to-stream fn :extra-stats - whether or not in depth latency statistics should be tracked on this subscription (false by default) :timeout-ms - the amount of time after which a message should be considered to be timedout and retried (milliseconds, 30000 by default) :max-retries - the maximum number of retries (due to timeout) before a message get considered to be parked (500 by default) :live-buf-sz - the size of the buffer listening to live messages as they happen (500 by default) :read-batch-sz - the number of events read at a time when paging in history (10 by default) :hist-buf-sz - the number of events to cache when paging through history (20 by default) :checkpoint-after-ms - the amount of time to try to checkpoint after (milliseconds, 2000 by default) :min-checkpoints - the minimum number of messages to checkpoint (10 by default) :max-checkpoints - the maximum number of messages to checkpoint. if this number is a reached a checkpoint will be forced. (1000 by default) :max-subscribers - the maximum number of subscribers allowed (default: 0) :strategy - the strategy to use for distributing events to client consumers: - :round-robin (default) distributes events to each client in a round robin fashion - :dispatch-to-single distributes events to a single client until it is full. then round robin to the next client.

Returns a deferred which derefs to :done on success. The failed return value - see write-events fn.

delete-persistent-subscription clj

(delete-persistent-subscription {:keys [conn stream group] :as m})

Asynchronously deletes a persistent subscription group on a stream. Please refer to create-persistent-subscription fn for more info.

delete-stream clj

(delete-stream {:keys [conn stream exp-ver hard-delete req-master]
                :or {hard-delete false req-master true}
                :as m})

Deletes a stream from the event store.

Options: :conn - (required) see write-events fn :stream - ” :exp-ver - ” :hard-delete - indicator for tombstoning vs soft-deleting the stream. tombstoned streams can never be recreated. soft-deleted streams can be written to again, but the event number sequence will not start from 0. (false by default) :req-master - see write-events fn :login - ” :password - ”

The successful return value is a deferred which derefs to a map with a key: :pos - see write-events fn The failed return value - see write-events fn.

deserialize clj multimethod

(deserialize bytes format)

Deserializes bytes to a format where bytes is a byte array and format is a keyword. Please refer to serialize multimethod for an info about formats.

disconnect clj

(disconnect conn)

Disconnects connection to the event store. Returns a deferred which derefs to :done when the system is terminated.

WARNING: uses a hack to get a private field from the JVM Client!

get-stream-metadata clj

(get-stream-metadata {:keys [conn stream format] :as m})

Reads the metadata for a stream.

Options: :conn - (required) see write-events fn :stream - ” :format - see write-events fn :login - ” :password - ”

The successful return value is a deferred derefing to deserialized metadata. The failed return value - see write-events fn.

manual-ack clj

(manual-ack sub event)

Sends an acknowledgement for an event. sub should be a return value of the persistently-subscribe fn. Returns nil.

persistently-subscribe clj

(persistently-subscribe
  {:keys [conn stream group auto-ack] :or {auto-ack true} :as m}
  {:keys [live event where error] :as callbacks})

Starts listening to a persistent subscription.

Persistent Subscriptions (extract from the Event Store documentation): This kind of subscriptions supports the “competing consumers” messaging pattern. The subscription state is stored server side in the Event Store and allows for at-least-once delivery guarantees across multiple consumers on the same stream. It is possible to have many groups of consumers compete on the same stream, with each group getting an at-least-once guarantee.

Options: :conn - (required) see write-events fn :stream - ” :group - (required) the name of a group which should be created with the create-persistent-subscription fn first :auto-ack - if this is true (the default), an acknowledgement will be sent automatically on each event. otherwise, use the manual-ack fn. :format - see write-events fn :meta-format - ” :login - ” :password - ”

Please refer to subscribe-to-stream fn for an info about callbacks noting that the :close callback does not exist here.

The return value - see subscribe-to-stream fn.

WARNING: uses a hack to get a private field from the JVM Client!

read-all-streams clj

(read-all-streams {:keys [conn pos resolve-link-tos req-master where]
                   :or {resolve-link-tos false req-master true}
                   :as m}
                  max-count)

Reads all events in the node forward (e.g. beginning to end) or backwards (e.g. end to beginning) starting from position.

Options: :conn - (required) see write-events fn :pos - position to start reading from. if omitted, reading starts at the first or at the latest event (depending on direction). for paging use :next-pos returned in metadata. :resolve-link-tos - see read-event fn :req-master - see write-events fn :format - ” :meta-format - ” :login - ” :password - ” :where - see subscribe-to-stream fn

If max-count is positive, direction is forward, otherwise is backward.

The successful return value is a deferred which derefs to a vector of events. Please refer to read-event fn for an info about returned events. One extra metadata returned with each event is: :pos - the event position (has :commit and :prepare subkeys) Additionally, the vector itself has Clojure metadata with values for: :dir - either :forward or :backward :this-pos - the position of this read (has :commit and :prepare subkeys) :next-pos - the next position to use for paging (has :commit and :prepare subkeys) The failed return value - see write-events fn.

read-event clj

(read-event {:keys [conn stream resolve-link-tos req-master]
             :or {resolve-link-tos false req-master true}
             :as m}
            num)

Reads a single event from a stream at event number num. If num is nil, reads the latest event.

Options: :conn - (required) see write-events fn :stream - ” :resolve-link-tos - whether to resolve LinkTo events automatically (false by default) :req-master - see write-events fn :format - ” :meta-format - ” :login - ” :password - ”

The successful return value is a deferred which derefs to a deserialized event which has Clojure metadata with values for: :id - the id of an event (UUID) :type - the type of an event (string) :num - the event number :stream - the stream name :date - an org.joda.time.DateTime when the event was added (can be nil) :meta - (optional) the deserialized event metadata If an event is not a clojure.lang.IObj, it will be wrapped like this: {:eventful.core/val <your event>} in order to be able to add Clojure metadata to it. You can use the unwrap convenience fn. The failed return value - see write-events fn.

Example: (let [conn (connect {:hostname "127.0.0.1" :port 1113 :login "admin" :password "changeit"})] (read-event {:conn conn :stream "inventory-item-1"} 0))

read-stream clj

(read-stream {:keys [conn stream resolve-link-tos req-master where]
              :or {resolve-link-tos false req-master true}
              :as m}
             [start max-count :as v])

Reads count events from a stream forwards (e.g. oldest to newest) or backwards (e.g. newest to oldest) starting from event number.

Options: :conn - (required) see write-events fn :stream - ” :resolve-link-tos - see read-event fn :req-master - see write-events fn :format - ” :meta-format - ” :login - ” :password - ” :where - see subscribe-to-stream fn

The second argument determines starting point, direction and count. If start is nil, reading starts at the first or at the latest event (depending on direction). If max-count is positive, direction is forward, otherwise is backward. Please refer to eventful.core-test namespace for examples.

The successful return value is a deferred which derefs to a vector of events. Please refer to read-event fn for an info about returned events. Additionally, the vector itself has Clojure metadata with values for: :event-num - a map with the :next event number (can be nil) and the :last event number :end-of-stream - a boolean :last-commit-pos - a number :dir - either :forward or :backward The failed return value - see write-events fn.

reduce-stream clj

(reduce-stream {:keys [init batch-sz] :or {batch-sz 500} :as m} f)

Reads all events from a stream forwards and reduces them using a fn f. Please refer to read-event fn for an info about events passed to f.

Options (in addition to all of the options accepted by read-stream fn): :init - initial value of reduction (nil by default) :batch-sz - batch size (number of events, 500 by default)

The successful return value is a deferred which derefs to a reduction value. Its metadata is merged with read-stream fn result metadata, so see unwrap fn. The failed return value - see write-events fn.

serialize clj multimethod

(serialize x format)

Serializes x to a format where format is a keyword.

Built-in formats: Transit - Transit-JSON is the default format :edn - EDN :json - cheshire dependency should be added to your project and eventful.json namespace should be required first :bytes - byte arrays

To add a custom format you can use eventful.json namespace as a starting point.

set-stream-metadata clj

(set-stream-metadata {:keys [conn stream exp-ver format] :as m} metadata)

Sets the metadata for a stream.

Options: :conn - (required) see write-events fn :stream - ” :exp-ver - ” :format - see write-events fn :login - ” :password - ”

The metadata can be anything (de)serializable.

The return value - see write-events fn. The difference is that :next-exp-ver can be nil.

Please note that the :exp-ver input and the :next-exp-ver output refer to the metadata stream as opposed to the stream itself.

subscribe-to-all-streams clj

(subscribe-to-all-streams
  {:keys [conn pos resolve-link-tos] :or {resolve-link-tos false} :as m}
  {:keys [live event where error close] :as callbacks})

Subscribes to all events in the event store. New events written to the stream while the subscription is active will be pushed to the client. If :pos is specified, existing events after position :pos (excluding) are read from the event store and presented to the user of callbacks as if they had been pushed. Once the end of the stream is read the subscription is transparently (to the user) switched to push new events as they are written. If events have already been received and resubscription from the same point is desired, use the position representing the last event processed which appeared on the subscription.

Options: :conn - (required) see write-events fn :pos - either an excluded position (a map with :commit and :prepare keys) to start reading after or :zero to read all events or omitted for live subscription only :resolve-link-tos - see read-event fn :format - see write-events fn :meta-format - ” :login - ” :password - ”

Please refer to subscribe-to-stream fn for an info about callbacks. One extra metadata received with each event is: :pos - the event position which can be used as the :pos option to resubscribe (has :commit and :prepare subkeys)

The return value - see subscribe-to-stream fn.

subscribe-to-stream clj

(subscribe-to-stream
  {:keys [conn stream from resolve-link-tos] :or {resolve-link-tos false} :as m}
  {:keys [live event where error close] :as callbacks})

Subscribes to a single event stream. New events written to the stream while the subscription is active will be pushed to the client. If :from is specified, existing events :from onwards are read from the stream and presented to the user of callbacks as if they had been pushed. Once the end of the stream is read the subscription is transparently (to the user) switched to push new events as they are written. If events have already been received and resubscription from the same point is desired, use the event number of the last event processed which appeared on the subscription (incremented by 1).

Options: :conn - (required) see write-events fn :stream - ” :from - the starting event number (if omitted, live subscription only) :resolve-link-tos - see read-event fn :format - see write-events fn :meta-format - ” :login - ” :password - ”

Callbacks: :live - called when subscription becomes live (no arguments) :event - (required) called with one argument which is the event received. please refer to read-event fn for an info about the received events. :where - called before :event with one argument which is the event metadata map. :event will not be called if the return value is falsy. please refer to read-event fn for an info about metadata map. the difference is that the value of :meta is bytes (use deserialize multimethod). :error - called with one argument which is the error map - see write-events fn :close - called when subscription closes (no arguments)

The return value can be used to close the subscription by calling the close-subscription fn or by using the Clojure with-open macro.

tx-commit clj

(tx-commit tx)

Commits this transaction.

Returns a deferred which derefs to :done on success. The failed return value - see write-events fn.

tx-cont clj

(tx-cont {:keys [conn] :as m} id)

Continues transaction by provided transaction id which can be obtained with the tx-id fn first.

Options: :conn - (required) see write-events fn :login - see write-events fn :password - ”

The successful return value is a future which derefs to a new transaction object. The failed return value - see write-events fn.

tx-id clj

(tx-id tx)

Gets id of a transaction tx returned by the tx-start or tx-cont fns.

tx-start clj

(tx-start {:keys [conn stream exp-ver req-master] :or {req-master true} :as m})

Starts a transaction in the event store on a given stream asynchronously. A transaction allows the calling of multiple writes with multiple round trips over long periods of time between the caller and the event store.

Options: :conn - (required) see write-events fn :stream - (required) the stream to start a transaction on, e.g. "inventory-item-1" :exp-ver - (required) the expected version of the stream at the time of starting the transaction, see write-events fn :req-master - see write-events fn :login - ” :password - ”

The successful return value is a deferred which derefs to a transaction object. The failed return value - see write-events fn.

tx-write-events clj

(tx-write-events {:keys [tx] :as m} & events)

Writes to a transaction in the event store asynchronously.

Options: :tx - the transaction returned by the tx-start or tx-cont fns :format - see write-events fn :meta-format - ”

Please refer to write-events fn for an info about events to write.

Returns a deferred which derefs to :done on success. The failed return value - see write-events fn.

unwrap clj

(unwrap val)

Reverse of wrap fn. See also read-event fn.

update-persistent-subscription clj

(update-persistent-subscription {:keys [conn stream group] :as m} settings)

Asynchronously updates a persistent subscription group on a stream. Please refer to create-persistent-subscription fn for more info.

wrap clj

(wrap val)

If val is not a clojure.lang.IObj, wraps it in order to be able to add Clojure metadata to it. See also write-events fn.

write-events clj

(write-events {:keys [conn stream exp-ver] :as m} & events)

Writes events to a stream.

Options: :conn - (required) connection returned by the connect fn :stream - (required) stream name, e.g. "inventory-item-1" :exp-ver - (required) expected version, which is either a number, :any or :no-stream - see notes section below :req-master - should the event store refuse operation if it is not master? (true by default) :format - event serialization format - please refer to serialize multimethod for an info about formats (Transit-JSON by default) :meta-format - event metadata serialization format (see :format above) :login - optional user credentials to perform operation with :password - ”

About expected version: When writing events to a stream the :exp-ver choice can make a very large difference in the observed behavior. For example, if no stream exists and :any is used, a new stream will be implicitly created when appending. There are also differences in idempotency between different types of calls. If you specify an expected version aside from :any the event store will give you an idempotency guarantee. If using :any the event store will do its best to provide idempotency but does not guarantee idempotency.

An event can be anything as long as it can be (de)serialized. Please refer to serialize and deserialize multimethods for an info about this. It can have Clojure metadata with optional values for: :type - the type of an event, "event" by default :id - the id of an event (UUID), random UUID by default :meta - the arbitrary event metadata stored in the event store - anything (de)serializable. If an event is not a clojure.lang.IObj and you want to add Clojure metadata to it, first please wrap it like this: {:eventful.core/val <your event>}. You can use the wrap convenience fn.

The return value is a deferred which derefs to a map with values for:

• in case of a success: :next-exp-ver - the next expected version for the stream :pos - the position of the write in the log (has :commit and :prepare subkeys, can be nil)
• in case of a failure (needs to be catched on deref and read via ex-data): :error-type - :wrong-exp-ver, :stream-not-found, :not-authenticated or :other :error - a Throwable

Example: (let [conn (connect {:hostname "127.0.0.1" :port 1113 :login "admin" :password "changeit"})] (write-events {:conn conn :stream "inventory-item-1" :exp-ver :no-stream} {:event :created :name "foo"}))