README
¶
Go.GetEventStore
A Golang client for EventStore 3.x HTTP API.
Go.GetEventStore is a http client for GetEventStore written in Go. The client abstracts interaction with the GetEventStore HTTP API providing easy to use features for reading and writing of events and event metadata.
CQRS reference implementation
An full example CQRS implementation using go.geteventstore can be found at go.cqrs
Supported features
| Feature | Description |
|---|---|
| Write Events & Event Metadata | Writing single and multiple events to a stream. Optionally expected version can be provided if you want to use optimistic concurrency features of the eventstore. |
| Read Events & Event Metadata | Reading events & event metadata from a stream. |
| Read & Write Stream Metadata | Read and writing stream metadata. |
| Basic Authentication | |
| Long Poll | Long Poll allows the client to listen at the head of a stream for new events. |
| Soft & Hard Delete Stream | |
| Catch Up Subscription | Using long poll with a StreamReader provides an effective catch up subscription. |
| Serialization & Deserialization of Events | The package handles serialization and deserialization of your application events to and from the eventstore. |
| Reading Stream Atom Feed | The package provides methods for reading stream Atom feed pages, returning a fully typed struct representation. |
| Setting Optional Headers | Optional headers can be added and removed. |
Below are some code examples giving a summary view of how the client works. To learn to use the client in more detail, heavily commented example code can be found in the examples directory.
Get the package
$ go get github.com/dkimot/go.geteventstore
Import the package
import "github.com/dkimot/go.geteventstore"
Create a new client
client, err := goes.NewClient(nil, "http://youreventstore:2113")
if err != nil {
log.Fatal(err)
}
Set basic authentication
If required, you can set authentication on the client. Credentials can be changed at any time. Requests are made with the credentials that were set last or none if none are set.
client.SetBasicAuth("admin", "changeit")
Write events and event Metadata
Writing events and event metadata are supported via the StreamWriter.
// Create your event
myEvent := &FooEvent{
FooField: "Lorem Ipsum",
BarField: "Dolor Sit Amet",
BazField: 42,
}
// Create your metadata type
myEventMeta := make(map[string]string)
myEventMeta["Foo"] = "consectetur adipiscing elit"
// Wrap your event and event metadata in a goes.Event
myGoesEvent := goes.NewEvent(goes.NewUUID(), "FooEvent", myEvent, myEventMeta)
// Create a new StreamWriter
writer := client.NewStreamWriter("FooStream")
// Write the event to the stream, here we pass nil as the expectedVersion as we
// are not wanting to flag concurrency errors
err := writer.Append(nil, myGoesEvent)
if err != nil {
// Handle errors
}
Read events
Reading events using the goes.StreamReader loosely follows the iterator idiom used in the "database/sql" package and other go libraries that deal with databases. This idiom works very well for reading events and provides an easy way to control the rate at which events are returned and to poll at the head of a stream.
An example of how to read all the events in a stream and then exit can be found in the read and write events example.
An example of how to read up to the head of the stream and then continue to listen for new events can be found in the longpoll example.
// Create a new goes.StreamReader
reader := client.NewStreamReader("FooStream")
// Call Next to get the next event
for reader.Next() {
// Check if the call resulted in an error.
if reader.Err() != nil {
// Handle errors
}
// If the call did not result in an error then an event was returned
// Create the application types to hold the deserialized even data and meta data
fooEvent := FooEvent{}
fooMeta := make(map[string]string)
// Call scan to deserialize the event data and meta data into your types
err := reader.Scan(&fooEvent, &fooMeta)
if err != nil {
// Handle errors that occured during deserialization
}
}
Long polling head of a stream
LongPoll provides an easy and efficient way to poll a stream listening for new events. The server will wait the specified amount of time or until new events are available on a stream.
reader := client.NewStreamReader("FooStream")
for reader.Next() {
if reader.Err() != nil {
// When there are no more event in the stream, set LongPoll.
// The server will wait for 15 seconds in this case or until
// events become available on the stream.
if e, ok := reader.Err().(*goes.ErrNoMoreEvents); ok {
reader.LongPoll(15)
}
} else {
fooEvent := FooEvent{}
_ := reader.Scan(&fooEvent, &fooMeta)
}
}
A more detailed example of using LongPoll can be found in the examples directory.
Deleting streams
The client supports both soft delete and hard delete of event streams.
// Soft delete or hard delete is specified by a boolean argument
// here foostream will be soft deleted.
resp, err := client.DeleteStream("foostream", false)
Example code for deleting streams can be found in the examples directory.
Direct use of the client
The StreamReader and StreamWriter types are the easiest way to read and write events. If you would like to implement some other logic around reading events, some methods are available on the Client type. An example is included demonstrating how to use some of these methods.
###Setting optional headers Most of the optional headers are are included implicitly such as ES-LongPoll, ES-ExpectedVersion & ES-HardDelete when using LongPoll on the StreamReader or when appending events or deleting streams. However should you wish to use any of the others the can be set explicitly on the client. The only other ones you might want to use are ES-ResolveLinkTo, ES-RequiresMaster or ES-TrustedAuth.
// Setting a header will mean all subsequent requests will include the header
client.SetHeader("ES-ResolveLinkTo", "false")
// Deleting a header means that it will not be included in any subsequent requests.
client.DeleteHeader("ES-ResolveLinkTo")
Running the Unit Tests
To keep the library lightweight and easy to use, I have tried not to have any dependencies on other packages. To use the package there are no dependencies, to run the unit tests however, the package does require some dependencies.
You will need the go.geteventstore.testfeed package.
$ go get github.com/dkimot/go.geteventstore.testfeed
You will also need the Check.V1 package which is used for assertions in unit tests.
$ go get gopkg.in/check.v1
After this you should be able to run the tests as normal for any Golang project with unit tests.
$ go test
Documentation
¶
Overview ¶
Package goes provides an abstraction over GetEventStore's HTTP API.
The package simplifies reading and writing events to the HTTP API.
The package provides a Client which contains foundation methods for connecting to and interacting with the eventstore.
The package also provides StreamReader and StreamWriter types which provide methods for reading and writing events and metadata.
Index ¶
- func NewUUID() string
- type Client
- func (c *Client) DeleteHeader(key string)
- func (c *Client) DeleteStream(streamName string, hardDelete bool) (*Response, error)
- func (c *Client) Do(req *http.Request, v io.Writer) (*Response, error)
- func (c *Client) GetEvent(url string) (*EventResponse, *Response, error)
- func (c *Client) GetFeedPath(stream, direction string, version int, pageSize int) (string, error)
- func (c *Client) GetMetadataURL(stream string) (string, *Response, error)
- func (c *Client) NewRequest(method, urlString string, body interface{}) (*http.Request, error)
- func (c *Client) NewStreamReader(streamName string) *StreamReader
- func (c *Client) NewStreamWriter(streamName string) *StreamWriter
- func (c *Client) NewVersionedStreamReader(streamName string, version int) *StreamReader
- func (c *Client) ReadFeed(url string) (*atom.Feed, *Response, error)
- func (c *Client) SetBasicAuth(username, password string)
- func (c *Client) SetHeader(key, value string)
- type ErrBadRequest
- type ErrConcurrencyViolation
- type ErrDeleted
- type ErrNoMoreEvents
- type ErrNotFound
- type ErrTemporarilyUnavailable
- type ErrUnauthorized
- type ErrUnexpected
- type ErrorResponse
- type Event
- type EventAtomResponse
- type EventResponse
- type Link
- type Response
- type StreamReader
- func (s *StreamReader) Err() error
- func (s *StreamReader) EventResponse() *EventResponse
- func (s *StreamReader) LongPoll(seconds int)
- func (s *StreamReader) MetaData() (*EventResponse, error)
- func (s *StreamReader) Next() bool
- func (s *StreamReader) NextVersion(version int)
- func (s *StreamReader) Scan(e interface{}, m interface{}) error
- func (s *StreamReader) Version() int
- type StreamWriter
- type TimeStr
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is a handle for an eventstore server.
The client is used to store connection details such as server URL, basic authentication credentials and headers.
The client also provides methods interacting with the eventstore.
In general, the StreamReader and StreamWriter should be used to interact with the eventstore. These methods further abstract methods on the client, however you can also directly use methods on the client to interact with the eventstore if you want to create some custom behaviour.
func NewClient ¶
NewClient returns a new client.
httpClient will usually be nil and the client will use the http.DefaultClient. Should you want to implement behaviours at the transport level you can provide your own *http.Client
serverURL is the full URL to your eventstore server including protocol scheme and port number.
func (*Client) DeleteHeader ¶
DeleteHeader deletes a header from the collection of headers.
func (*Client) DeleteStream ¶
DeleteStream will delete a stream
Streams may be soft deleted or hard deleted.
Soft deleting a stream means that you can later recreate it simply by appending events to it.
Hard deleting a stream means that it has permanently been deleted and can never be recreated.
http://docs.geteventstore.com/http-api/3.8.0/deleting-a-stream/
func (*Client) Do ¶
Do executes requests to the server.
The response body is copied into v if v is not nil. Returns a *Response that wraps the http.Response returned from the server. The response body is available in the *Response in case the consumer wishes to process it in some way rather than read if from the argument v
func (*Client) GetEvent ¶
func (c *Client) GetEvent(url string) (*EventResponse, *Response, error)
GetEvent reads a single event from the eventstore.
The event response will be nil in an error case. *Response may be nil if an error occurs before the http request. Otherwise it will contain the raw http response and status. If an error occurs during the http request an *ErrorResponse will be returned as the error. The *ErrorResponse will contain the raw http response and status and a description of the error.
func (*Client) GetFeedPath ¶
GetFeedPath returns the path for a feedpage
Valid directions are "forward" and "backward".
To get the path to the head of the stream, pass a negative integer in the version argument and "backward" as the direction.
func (*Client) GetMetadataURL ¶
GetMetadataURL gets the url for the stream metadata. according to the documentation the metadata url should be acquired through a query to the stream feed as the authors of GetEventStore reserve the right to change the url. http://docs.geteventstore.com/http-api/latest/stream-metadata/
func (*Client) NewRequest ¶
NewRequest creates a new *http.Request that can be used to execute requests to the server using the client.
func (*Client) NewStreamReader ¶
func (c *Client) NewStreamReader(streamName string) *StreamReader
NewStreamReader returns a new *StreamReader.
func (*Client) NewStreamWriter ¶
func (c *Client) NewStreamWriter(streamName string) *StreamWriter
NewStreamWriter returns a new *StreamWriter.
func (*Client) NewVersionedStreamReader ¶ added in v1.0.1
func (c *Client) NewVersionedStreamReader(streamName string, version int) *StreamReader
NewVersionedStreamReader returns a new *StreamReader that starts reading after the specified version
func (*Client) ReadFeed ¶
ReadFeed reads the atom feed for a stream and returns an *atom.Feed.
The feed object returned may be nil in case of an error. The *Response may also be nil if the error occurred before the http request. If the error occurred after the http request, the *Response will contain the raw http response and status. If the error occurred during the http request an *ErrorResponse will be returned and this will also contain the raw http request and status and an error message.
func (*Client) SetBasicAuth ¶
SetBasicAuth sets the credentials for requests.
Credentials will be read from the client before each request.
type ErrBadRequest ¶
type ErrBadRequest struct {
ErrorResponse *ErrorResponse
}
ErrBadRequest is returned when the server returns a bad request error
func (ErrBadRequest) Error ¶
func (e ErrBadRequest) Error() string
type ErrConcurrencyViolation ¶
type ErrConcurrencyViolation struct {
ErrorResponse *ErrorResponse
}
ErrConcurrencyViolation is returned when the expected version does not match the stream version when writing to an event stream.
func (ErrConcurrencyViolation) Error ¶
func (e ErrConcurrencyViolation) Error() string
type ErrDeleted ¶
type ErrDeleted struct {
ErrorResponse *ErrorResponse
}
ErrDeleted is returned when a request is made to a stream that has been hard deleted.
func (ErrDeleted) Error ¶
func (e ErrDeleted) Error() string
type ErrNoMoreEvents ¶
type ErrNoMoreEvents struct{}
ErrNoMoreEvents is returned when there are no events to return from a request to a stream.
func (ErrNoMoreEvents) Error ¶
func (e ErrNoMoreEvents) Error() string
type ErrNotFound ¶
type ErrNotFound struct {
ErrorResponse *ErrorResponse
}
ErrNotFound is returned when a stream is not found.
func (ErrNotFound) Error ¶
func (e ErrNotFound) Error() string
type ErrTemporarilyUnavailable ¶
type ErrTemporarilyUnavailable struct {
}
ErrTemporarilyUnavailable is returned when the server returns ServiceUnavailable.
This error may be returned if a request is made to the server during startup. When the server starts up initially and the client is completely unable to connect to the server a *url.Error will be returned. Once the server is up but not ready to serve requests a ServiceUnavailable error will be returned for a brief period.
func (ErrTemporarilyUnavailable) Error ¶
func (e ErrTemporarilyUnavailable) Error() string
type ErrUnauthorized ¶
type ErrUnauthorized struct {
}
ErrUnauthorized is returned when a request to the eventstore is not authorized
func (ErrUnauthorized) Error ¶
func (e ErrUnauthorized) Error() string
type ErrUnexpected ¶
type ErrUnexpected struct {
ErrorResponse *ErrorResponse
}
ErrUnexpected is returned when a request to the eventstore returns an error that is not explicitly represented by a goes Error type such as UnauthorisedError or ErrNotFound
func (ErrUnexpected) Error ¶
func (e ErrUnexpected) Error() string
type ErrorResponse ¶
ErrorResponse encapsulates data about an interaction with the eventstore that produced an HTTP error.
An ErrorResponse embeds the raw *http.Response and provides access to the raw http.Request that resulted in an error. Status contains the status message returned from the server. StatusCode contains the status code returned from the server.
func (*ErrorResponse) Error ¶
func (r *ErrorResponse) Error() string
type Event ¶
type Event struct {
EventStreamID string `json:"eventStreamId,omitempty"`
EventNumber int `json:"eventNumber,omitempty"`
EventType string `json:"eventType,omitempty"`
EventID string `json:"eventId,omitempty"`
Data interface{} `json:"data"`
Links []Link `json:"links,omitempty"`
MetaData interface{} `json:"metadata,omitempty"`
}
Event encapsulates the data of an eventstore event.
EventStreamID is the id returned in the event atom response. EventNumber represents the stream version for this event. EventType describes the event type. EventID is the guid of the event. Data contains the data of the event. Links contains the urls of the event on the evenstore MetaData contains the metadata for the event.
func NewEvent ¶
NewEvent creates a new event object.
If an empty eventId is provided a new uuid will be generated automatically and retured in the event. If an empty eventType is provided the eventType will be set to the name of the type provided. data and meta can be nil.
func (*Event) PrettyPrint ¶
PrettyPrint renders an indented json view of the Event object.
type EventAtomResponse ¶
type EventAtomResponse struct {
Title string `json:"title"`
ID string `json:"id"`
Updated TimeStr `json:"updated"`
Summary string `json:"summary"`
Content interface{} `json:"content"`
}
EventAtomResponse is used internally to unmarshall the raw response
func (*EventAtomResponse) PrettyPrint ¶
func (e *EventAtomResponse) PrettyPrint() string
PrettyPrint renders and indented json view of the eventAtomResponse
type EventResponse ¶
EventResponse encapsulates the response for an event reflecting the atom response returned from the server which contains data in addition to the actual event when requested as content type application/vnd.eventstore.atom+json
For more information on the server response see: http://docs.geteventstore.com/http-api/latest/reading-streams/
func (*EventResponse) PrettyPrint ¶
func (e *EventResponse) PrettyPrint() string
PrettyPrint renders an indented json view of the EventResponse.
type Response ¶
Response encapsulates HTTP responses from the server.
A Response object contains the raw http response, the status code returned and the status message returned.
A Response object is returned from all methods on the client that interact with the the eventstore. It is intended to provide access to data about the response in case the user wants to inspect the response such as the status or the raw http response.
type StreamReader ¶
type StreamReader struct {
// contains filtered or unexported fields
}
StreamReader provides methods for reading events and event metadata.
func (*StreamReader) Err ¶
func (s *StreamReader) Err() error
Err returns any error that is raised as a result of a call to Next().
func (*StreamReader) EventResponse ¶
func (s *StreamReader) EventResponse() *EventResponse
EventResponse returns the container for the event that is returned from a call to Next().
func (*StreamReader) LongPoll ¶
func (s *StreamReader) LongPoll(seconds int)
LongPoll causes the server to wait up to the number of seconds specified for results to become available at the URL requested.
LongPoll is useful when polling the end of the stream as it returns quickly after new events are found which means that it is more responsive than polling over some arbitrary time interval. It also reduces the number of unproductive calls to the server polling for events when there are no new events to return.
Setting the argument seconds to any integer value above 0 will cause the request to be made with ES-LongPoll set to that value. Any value 0 or below will cause the request to be made without ES-LongPoll and the server will not wait to return.
func (*StreamReader) MetaData ¶
func (s *StreamReader) MetaData() (*EventResponse, error)
MetaData gets the metadata for a stream.
Stream metadata is retured as an EventResponse.
For more information on stream metadata see: http://docs.geteventstore.com/http-api/3.7.0/stream-metadata/
func (*StreamReader) Next ¶
func (s *StreamReader) Next() bool
Next gets the next event on the stream.
Next should be treated more like a cursor over the stream rather than an enumerator over a collection of results. Individual events are retrieved on each call to Next().
The boolean returned is intended to provide a convenient mechanism to to enumerate and process events, it should not be considered an indication of the status of a call to Next(). To undertand the outcomes of operations the stream's Err() field should be inspected. It is left to the user to determine under what conditions to exit the loop.
When next is called, it will go to the eventstore and get a single event at the current reader's stream version.
func (*StreamReader) NextVersion ¶
func (s *StreamReader) NextVersion(version int)
NextVersion is the version of the stream that will be returned by a call to Next().
func (*StreamReader) Scan ¶
func (s *StreamReader) Scan(e interface{}, m interface{}) error
Scan deserializes event and event metadata into the types passed in as arguments e and m.
func (*StreamReader) Version ¶
func (s *StreamReader) Version() int
Version returns the current stream version of the reader.
type StreamWriter ¶
type StreamWriter struct {
// contains filtered or unexported fields
}
StreamWriter provides methods for writing events and metadata to an event stream.
func (*StreamWriter) Append ¶
func (s *StreamWriter) Append(expectedVersion *int, events ...*Event) error
Append writes an event to the head of the stream.
If the stream does not exist, it will be created.
There are some special version numbers that can be provided. http://docs.geteventstore.com/http-api/3.7.0/writing-to-a-stream/
-2 : The write should never conflict with anything and should always succeed.
-1 : The stream should not exist at the time of writing. This write will create it.
0 : The stream should exist but it should be empty.
func (*StreamWriter) WriteMetaData ¶
func (s *StreamWriter) WriteMetaData(stream string, metadata interface{}) error
WriteMetaData writes the metadata for a stream.
The operation will replace the current stream metadata.
For more information on stream metadata see: http://docs.geteventstore.com/http-api/3.7.0/stream-metadata/
If the metadata was written successfully the error returned will be nil.
If an error occurs the error returned may be an ErrUnauthorized, a ErrTemporarilyUnavailable or an ErrUnexpected if the error occurred during a http request to the server. In these cases, the *ErrorResponse will be available for inspection as an ErrorResponse field on the error. If an error occurred outside of the http request another type of error will be returned such as a *url.Error in cases where the streamwriter is unable to connect to the server.
Source Files
Directories