event_streams

Example of appending an event to an existing stream with WriteStreamRevisionAny.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"

	"github.com/google/uuid"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	firstEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// Create a stream by appending one event to it
	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{firstEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}

	// Append an event to an existing stream
	secondEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}
	writeResult, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionAny{},
		[]event_streams.ProposedEvent{secondEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 1 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}
}

Output:

Example of appending an event to a stream which does not exist with WriteStreamRevisionAny.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"

	"github.com/google/uuid"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"
	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionAny{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}
}

Output:

Example of appending an event to an existing stream with exact expected stream revision.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"

	"github.com/google/uuid"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	firstEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// Create a stream by appending one event to it
	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{firstEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}

	// Append an event to an existing stream
	secondEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}
	writeResult, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevision{Revision: 0}, // 0 because stream has one event, like an index in a slice
		[]event_streams.ProposedEvent{secondEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 1 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}
}

Output:

Example of appending an event to a stream which does not exist with WriteStreamRevisionNoStream.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"

	"github.com/google/uuid"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"
	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln(writeResult.GetCurrentRevision())
	}
}

Output:

Example demonstrates how to do a batch append to a stream which does not exist. Correlation id for events will auto be generated.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)
	streamId := "some_stream"

	firstEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	deadline := time.Now().Add(time.Second * 10)
	// batch append to a stream which does not exist
	writeResult, err := client.BatchAppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{firstEvent},
		1,
		deadline)

	if writeResult.IsCurrentRevisionNoStream() {
		log.Fatalln("IsCurrentRevisionNoStream should return false")
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln("Current revision should be 0")
	}
}

Output:

Example demonstrates how to do a batch append to a stream which does exist. Correlation id for events will auto be generated.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)
	streamId := "some_stream"

	firstEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream by appending one event to it
	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{firstEvent})
	if err != nil {
		log.Fatalln(err)
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln("Current revision must be 0")
	}

	secondEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	deadline := time.Now().Add(time.Second * 10)
	// batch append to a stream which exists with expected revision
	batchWriteResult, err := client.BatchAppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevision{Revision: 0}, // there is already one event in the stream
		[]event_streams.ProposedEvent{secondEvent},
		1,
		deadline)
	if err != nil {
		log.Fatalln(err)
	}

	if batchWriteResult.IsCurrentRevisionNoStream() {
		log.Fatalln("IsCurrentRevisionNoStream should return false")
	}

	if batchWriteResult.GetCurrentRevision() != 0 {
		log.Fatalln("Current revision should be 0")
	}
}

Output:

Example demonstrates how to do a batch append to a stream with correlation id.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)
	streamId := "some_stream"

	firstEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	deadline := time.Now().Add(time.Second * 10)
	correlationId := uuid.New()
	// batch append to a stream with correlation id
	writeResult, err := client.BatchAppendToStreamWithCorrelationId(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		correlationId,
		[]event_streams.ProposedEvent{firstEvent},
		1,
		deadline)

	if writeResult.IsCurrentRevisionNoStream() {
		log.Fatalln("IsCurrentRevisionNoStream should return false")
	}

	if writeResult.GetCurrentRevision() != 0 {
		log.Fatalln("Current revision should be 0")
	}
}

Output:

Example of soft-deleting a stream which does not exist.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	// delete a non-existing stream
	deleteResult, err := client.DeleteStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{})
	deletePosition, isPosition := deleteResult.GetPosition()

	// result of a soft-delete must be a position
	if !isPosition {
		log.Fatalln("Must be a position")
	}

	// position returned by soft-delete must not be zero
	if deletePosition.CommitPosition == 0 || deletePosition.PreparePosition == 0 {
		log.Fatalln("Commit and Prepare position must not be zero")
	}
}

Output:

Example of sot-deleting a stream which exists.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"
	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream with one event
	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	// soft-delete a stream
	deleteResult, err := client.DeleteStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevision{Revision: writeResult.GetCurrentRevision()})

	deletePosition, isPosition := deleteResult.GetPosition()

	if !isPosition {
		log.Fatalln("Must be a position")
	}

	writePosition, _ := writeResult.GetPosition()

	// position returned by soft-delete must be greater than the one of the last event
	if !deletePosition.GreaterThan(writePosition) {
		log.Fatalln("Delete position must be greater than last event's write position")
	}
}

Output:

Example of reading metadata for a stream which has no metadata set.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream with one event
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	// read stream's metadata
	metaDataResponse, err := client.GetStreamMetadata(context.Background(), streamId)
	if err != nil {
		log.Fatalln(err)
	}

	// Stream's metadata stream must not contains any metadata
	if !metaDataResponse.IsEmpty() {
		log.Fatalln("Stream's must have no metadata")
	}
}

Output:

Example of reading metadata for a stream which has metadata.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/ptr"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	expectedStreamMetadata := event_streams.StreamMetadata{
		MaxCount:              ptr.Int(17),
		TruncateBefore:        ptr.UInt64(10),
		CacheControlInSeconds: ptr.UInt64(17),
		MaxAgeInSeconds:       ptr.UInt64(15),
	}

	// write metadata for a stream
	_, err = client.SetStreamMetadata(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		expectedStreamMetadata,
	)
	if err != nil {
		log.Fatalln(err)
	}

	// read metadata for a stream
	metaData, err := client.GetStreamMetadata(context.Background(), streamId)
	if err != nil {
		log.Fatalln(err)
	}

	if metaData.IsEmpty() {
		log.Fatalln("Stream must have metadata")
	}

	if metaData.GetMetaStreamRevision() != 0 {
		log.Fatalln("Metadata must be at index 0 in stream's metadata stream")
	}

	if !reflect.DeepEqual(expectedStreamMetadata, metaData.GetStreamMetadata()) {
		log.Fatalln("Metadata received must be the same as the metadata written")
	}
}

Output:

Example of reading events backwards from the end of a stream $all. We will append some user defined events to demonstrate that user events will be read from the end of stream $all in reversed order.

That does not guarantee that system events are not going to be appended after user events if some system operation is triggered.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	// create 10 events to write (EventId must be unique)
	eventsToWrite := make(event_streams.ProposedEventList, 10)
	for i := uint32(0); i < 10; i++ {
		eventsToWrite[i] = event_streams.ProposedEvent{
			EventId:      uuid.Must(uuid.NewRandom()),
			EventType:    "TestEvent",
			ContentType:  "application/octet-stream",
			UserMetadata: []byte{},
			Data:         []byte("some event data"),
		}
	}

	// create a stream with 10 events
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		eventsToWrite)
	if err != nil {
		log.Fatalln(err)
	}

	count := uint64(len(eventsToWrite))

	readEvents, err := client.ReadEventsFromStreamAll(context.Background(),
		event_streams.ReadDirectionForward,
		stream_revision.ReadPositionAllStart{},
		count,
		false)
	if err != nil {
		log.Fatalln(err)
	}

	// Number of events read must equal to count
	if uint64(len(readEvents)) != count {
		log.Fatalln(`Number of events read from stream $all must 
					be greater than number of user defined events`)
	}

	// since events are read backwards from the end they are received in reversed order
	if !reflect.DeepEqual(eventsToWrite, readEvents.Reverse().ToProposedEvents()) {
		log.Fatalln("Events read from the end must match user defined events")
	}
}

Output:

Example of reading events from start of stream $all. At the beginning of stream $all we can find system events.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	// read events from stream $all
	// at the beginning of stream $all are some system events
	// as admin user we can access them too
	readEvents, err := client.ReadEventsFromStreamAll(context.Background(),
		event_streams.ReadDirectionForward,
		stream_revision.ReadPositionAllStart{},
		5,
		false)
	if err != nil {
		log.Fatalln(err)
	}

	if len(readEvents) < 5 {
		log.Fatalln("Not enough system events read from stream $all")
	}
}

Output:

Example of reading events backwards from the end of a stream.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	// create 10 events to write (EventId must be unique)
	eventsToWrite := make(event_streams.ProposedEventList, 10)
	for i := uint32(0); i < 10; i++ {
		eventsToWrite[i] = event_streams.ProposedEvent{
			EventId:      uuid.Must(uuid.NewRandom()),
			EventType:    "TestEvent",
			ContentType:  "application/octet-stream",
			UserMetadata: []byte{},
			Data:         []byte("some event data"),
		}
	}

	// create a stream with 10 events
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		eventsToWrite)
	if err != nil {
		log.Fatalln(err)
	}

	// read events from existing stream
	readEvents, err := client.ReadStreamEvents(context.Background(),
		streamId,
		event_streams.ReadDirectionBackward,
		stream_revision.ReadStreamRevisionEnd{},
		10, // set to be bigger than current number of events in a stream
		false)
	if err != nil {
		log.Fatalln(err)
	}

	// Event read must be in reversed order
	// since readEvents are of type ResolvedEvent we must convert them to slice of ProposedEvents
	if !reflect.DeepEqual(eventsToWrite, readEvents.Reverse().ToProposedEvents()) {
		log.Fatalln("Events read from a stream must match")
	}
}

Output:

Example of reading events from the start of a stream.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"
	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	eventsToWrite := []event_streams.ProposedEvent{proposedEvent}

	// create a stream with one event
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		eventsToWrite)
	if err != nil {
		log.Fatalln(err)
	}

	// read events from existing stream
	readEvents, err := client.ReadStreamEvents(context.Background(),
		streamId,
		event_streams.ReadDirectionForward,
		stream_revision.ReadStreamRevisionStart{},
		10, // set to be bigger than current number of events in a stream
		false)
	if err != nil {
		log.Fatalln(err)
	}

	// since readEvents are of type ResolvedEvent we must convert them to slice of ProposedEvents
	if !reflect.DeepEqual(eventsToWrite, readEvents.ToProposedEvents()) {
		log.Fatalln("Events read from a stream must match")
	}
}

Output:

Example of trying to read events from a stream which does not exist.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/errors"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, stdErr := connection.ParseConnectionString(clientURI)
	if stdErr != nil {
		log.Fatalln(stdErr)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	_, err := client.ReadStreamEvents(context.Background(),
		streamId,
		event_streams.ReadDirectionBackward,
		stream_revision.ReadStreamRevisionEnd{},
		1,
		false)

	if err.Code() != errors.StreamNotFoundErr {
		log.Fatalln("Stream must not exist")
	}
}

Output:

Example of trying to read events from a stream which is soft-deleted.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/errors"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, stdErr := connection.ParseConnectionString(clientURI)
	if stdErr != nil {
		log.Fatalln(stdErr)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream with one event
	writeResult, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	// soft-delete a stream
	_, err = client.DeleteStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevision{Revision: writeResult.GetCurrentRevision()})
	if err != nil {
		log.Fatalln(err)
	}

	// reading a soft-deleted stream fails with error code StreamNotFoundErr
	_, err = client.ReadStreamEvents(context.Background(),
		streamId,
		event_streams.ReadDirectionBackward,
		stream_revision.ReadStreamRevisionEnd{},
		event_streams.ReadCountMax,
		false)

	if err.Code() != errors.StreamNotFoundErr {
		log.Fatalln("Stream must not exist")
	}
}

Output:

Example of setting metadata for a stream which does not exist.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/ptr"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	expectedStreamMetadata := event_streams.StreamMetadata{
		MaxCount:              ptr.Int(17),
		TruncateBefore:        ptr.UInt64(10),
		CacheControlInSeconds: ptr.UInt64(17),
		MaxAgeInSeconds:       ptr.UInt64(15),
	}

	// write metadata for a stream
	_, err = client.SetStreamMetadata(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		expectedStreamMetadata,
	)
	if err != nil {
		log.Fatalln(err)
	}

	// read metadata for a stream
	metaData, err := client.GetStreamMetadata(context.Background(), streamId)
	if err != nil {
		log.Fatalln(err)
	}

	if metaData.IsEmpty() {
		log.Fatalln("Stream must have metadata")
	}

	if metaData.GetMetaStreamRevision() != 0 {
		log.Fatalln("Metadata must be at index 0 in stream's metadata stream")
	}

	if !reflect.DeepEqual(expectedStreamMetadata, metaData.GetStreamMetadata()) {
		log.Fatalln("Metadata received must be the same as the metadata written")
	}
}

Output:

Example of setting metadata for an existing stream.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/ptr"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream with one event
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	expectedStreamMetadata := event_streams.StreamMetadata{
		MaxCount:              ptr.Int(17),
		TruncateBefore:        ptr.UInt64(10),
		CacheControlInSeconds: ptr.UInt64(17),
		MaxAgeInSeconds:       ptr.UInt64(15),
	}

	// write metadata for a stream
	_, err = client.SetStreamMetadata(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		expectedStreamMetadata,
	)
	if err != nil {
		log.Fatalln(err)
	}

	// read stream's metadata
	metaData, err := client.GetStreamMetadata(context.Background(), streamId)
	if err != nil {
		log.Fatalln(err)
	}

	if metaData.IsEmpty() {
		log.Fatalln("Stream must have metadata")
	}

	if metaData.GetMetaStreamRevision() != 0 {
		log.Fatalln("Metadata must be at index 0 in stream's metadata stream")
	}

	if !reflect.DeepEqual(expectedStreamMetadata, metaData.GetStreamMetadata()) {
		log.Fatalln("Metadata received must be the same as the metadata written")
	}
}

Output:

Example demonstrates how to subscribe to stream $all with filter.

We create three streams and write events to them. Subscription to stream $all with a filter which will filter only content from two of the three streams. Content is filtered by prefix of the stream's ID.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/core/systemmetadata"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, stdErr := connection.ParseConnectionString(clientURI)
	if stdErr != nil {
		log.Fatalln(stdErr)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	prefix1 := "my_first_prefix"
	prefix2 := "my_second_prefix"

	otherStream := "read_all_existing_and_new_ones_otherStream"
	prefixStream := prefix1 + "_stream"
	newPrefixStream := prefix2 + "_stream"

	createEvents := func(count uint32) event_streams.ProposedEventList {
		result := make(event_streams.ProposedEventList, 10)
		for i := uint32(0); i < count; i++ {
			result[i] = event_streams.ProposedEvent{
				EventId:      uuid.Must(uuid.NewRandom()),
				EventType:    "TestEvent",
				ContentType:  "application/octet-stream",
				UserMetadata: []byte{},
				Data:         []byte("some event data"),
			}
		}
		return result
	}
	otherStreamEvents := createEvents(10)
	prefixStreamEvents := createEvents(10)
	newPrefixStreamEvents := createEvents(10)

	// create other stream with 10 events
	_, err := client.AppendToStream(context.Background(),
		otherStream,
		stream_revision.WriteStreamRevisionNoStream{},
		otherStreamEvents)
	if err != nil {
		log.Fatalln(err)
	}

	// create first stream which content we will read
	_, err = client.AppendToStream(context.Background(),
		prefixStream,
		stream_revision.WriteStreamRevisionNoStream{},
		prefixStreamEvents)
	if err != nil {
		log.Fatalln(err)
	}

	// subscribe to stream $all and filter only events written to
	// streams with prefix my_first_prefix and my_second_prefix
	streamReader, err := client.SubscribeToFilteredStreamAll(context.Background(),
		stream_revision.ReadPositionAllStart{},
		false,
		event_streams.Filter{
			FilterBy: event_streams.FilterByStreamId{
				Matcher: event_streams.PrefixFilterMatcher{
					PrefixList: []string{prefix1, prefix2},
				},
			},
			Window:                       event_streams.FilterNoWindow{},
			CheckpointIntervalMultiplier: 5,
		})
	if err != nil {
		log.Fatalln(err)
	}

	waitForReadingFirstEvents := sync.WaitGroup{}
	waitForReadingFirstEvents.Add(1)

	// read events written to a stream with prefix my_first_prefix
	go func() {
		defer waitForReadingFirstEvents.Done()

		var result event_streams.ProposedEventList
		readResult, err := streamReader.ReadOne()
		if err != nil {
			log.Fatalln(err)
		}

		if event, isEvent := readResult.GetEvent(); isEvent {
			result = append(result, event.ToProposedEvent())
		}

		if reflect.DeepEqual(prefixStreamEvents, result) {
			return
		}
	}()

	waitForNewEventsAppend := sync.WaitGroup{}
	waitForNewEventsAppend.Add(1)

	// after events from stream with prefix my_first_prefix are read
	// create stream with prefix my_second_prefix
	go func() {
		defer waitForNewEventsAppend.Done()
		waitForReadingFirstEvents.Wait() // wait until all events from stream with prefix my_first_prefix are read

		// create stream with prefix my_second_prefix with 10 events in it
		_, err = client.AppendToStream(context.Background(),
			newPrefixStream,
			stream_revision.WriteStreamRevisionNoStream{},
			newPrefixStreamEvents)
		if err != nil {
			log.Fatalln(err)
		}
	}()

	waitForReadingNewEvents := sync.WaitGroup{}
	waitForReadingNewEvents.Add(1)

	// read events written to a stream with prefix my_second_prefix
	go func() {
		defer waitForReadingNewEvents.Done()
		waitForNewEventsAppend.Wait() // wait until stream my_second_prefix created
		var result event_streams.ProposedEventList

		readResult, err := streamReader.ReadOne()
		if err != nil {
			log.Fatalln(err)
		}

		if event, isEvent := readResult.GetEvent(); isEvent {
			if !systemmetadata.IsSystemStream(event.Event.StreamId) {
				result = append(result, event.ToProposedEvent())
			}
		}

		// we have finished reading
		if reflect.DeepEqual(newPrefixStreamEvents, result) {
			return
		}
	}()
	// wait for reader to receive new events
	waitForReadingNewEvents.Wait()
}

Output:

Example demonstrates that subscription will receive StreamDeleted error once stream has been deleted.

package main

import (
	"context"
	"fmt"
	"log"
	"sync"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/errors"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	wg := sync.WaitGroup{}
	wg.Add(1)

	ctx := context.Background()

	// create a subscription to a stream, from start of the stream
	streamReader, err := client.SubscribeToStream(ctx,
		streamId,
		stream_revision.ReadStreamRevisionStart{},
		false)
	if err != nil {
		log.Fatalln(err)
	}

	go func() {
		defer wg.Done()
		_, err := streamReader.ReadOne() // reads content of a stream until StreamDeletedErr is received

		if err.Code() != errors.StreamDeletedErr {
			log.Fatalln("Unexpected error received")
		}
	}()

	_, err = client.TombstoneStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{})
	if err != nil {
		log.Fatalln(err)
	}

	// wait for reader to receive StreamDeleted
	wg.Wait()
}

Output:

Example shows that subscription from start of the stream will receive all old events written to it, as well as new events written to a stream after a subscription was created.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/errors"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, stdErr := connection.ParseConnectionString(clientURI)
	if stdErr != nil {
		log.Fatalln(stdErr)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	readerWait := sync.WaitGroup{}
	readerWait.Add(1)

	cancelWait := sync.WaitGroup{}
	cancelWait.Add(1)

	ctx := context.Background()
	ctx, cancelFunc := context.WithTimeout(ctx, 20*time.Second)

	createEvents := func(eventCount uint32) event_streams.ProposedEventList {
		result := make(event_streams.ProposedEventList, eventCount)

		for i := uint32(0); i < eventCount; i++ {
			result = append(result, event_streams.ProposedEvent{
				EventId:      uuid.Must(uuid.NewRandom()),
				EventType:    "TestEvent",
				ContentType:  "application/octet-stream",
				UserMetadata: []byte{},
				Data:         []byte("some event data"),
			})
		}
		return result
	}
	beforeEvents := createEvents(3)
	afterEvents := createEvents(2)

	totalEvents := append(beforeEvents, afterEvents...)

	// create a stream with 3 events in it
	_, err := client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		beforeEvents)
	if err != nil {
		log.Fatalln(err)
	}

	// create a stream subscription from start of the stream
	streamReader, err := client.SubscribeToStream(ctx,
		streamId,
		stream_revision.ReadStreamRevisionStart{},
		false)
	if err != nil {
		log.Fatalln(err)
	}

	go func() {
		defer readerWait.Done()

		var result event_streams.ProposedEventList

		for {
			response, err := streamReader.ReadOne() // read event one by-one
			if err != nil {
				if err.Code() == errors.CanceledErr { // we have received cancellation of a subscription
					break
				}
				cancelWait.Done() // must never be reached, some other error occur
				log.Fatalln("Unexpected error received")
			}

			event, isEvent := response.GetEvent()

			if !isEvent {
				log.Fatalln("Must have read an event")
			}
			result = append(result, event.ToProposedEvent())
			if len(result) == len(totalEvents) {
				cancelWait.Done() // we have read all events, signal to main thread that it can cancel subscription
			}
		}

		if !reflect.DeepEqual(totalEvents, result) {
			log.Fatalln("Not all events have been read from the stream")
		}
	}()

	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionAny{},
		afterEvents)

	cancelWait.Wait() // wait until subscription receives all events
	cancelFunc()      // cancel subscription

	readerWait.Wait() // wait for subscription go routine to exit
}

Output:

Example show how to subscribe to a stream which does not exist and wait for an event from it. We can subscribe to a non-existing stream. ReadOne method of StreamReader will block until stream with content is created.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	wg := sync.WaitGroup{}
	wg.Add(1)

	// subscribe to a stream
	// we can subscribe to a stream which does not exist
	// as soon as stream is created we will start to receive content from it.
	streamReader, err := client.SubscribeToStream(context.Background(),
		streamId,
		stream_revision.ReadStreamRevisionStart{},
		false)
	if err != nil {
		log.Fatalln(err)
	}

	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// we will wait for events in a separate go routine
	go func() {
		defer wg.Done()

		response, err := streamReader.ReadOne() // read blocks until event is written to a stream
		if err != nil {
			log.Fatalln(err)
		}

		event, isEvent := response.GetEvent()
		if !isEvent {
			log.Fatalln("Must have received an event")
		}

		if !reflect.DeepEqual(proposedEvent, event.ToProposedEvent()) {
			log.Fatalln("Must receive an event we have written to a stream")
		}
	}()

	// create a stream with one event written to it
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		event_streams.ProposedEventList{proposedEvent})

	wg.Wait()
}

Output:

Example shows how to subscribe to an existing stream from start.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamId := "some_stream"

	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}
	_, err = client.AppendToStream(context.Background(),
		streamId,
		stream_revision.WriteStreamRevisionNoStream{},
		event_streams.ProposedEventList{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	// subscribe to a stream fro start
	streamReader, err := client.SubscribeToStream(context.Background(),
		streamId,
		stream_revision.ReadStreamRevisionStart{},
		false)
	if err != nil {
		log.Fatalln(err)
	}

	wg := sync.WaitGroup{}
	wg.Add(1)
	// we will wait for an event in a separate go routine
	go func() {
		defer wg.Done()

		response, err := streamReader.ReadOne() // read an event written to a stream
		if err != nil {
			log.Fatalln(err)
		}

		event, isEvent := response.GetEvent()
		if !isEvent {
			log.Fatalln("Must have received an event")
		}

		if !reflect.DeepEqual(proposedEvent, event.ToProposedEvent()) {
			log.Fatalln("Must receive an event we have written to a stream")
		}
	}()

	wg.Wait()
}

Output:

Example demonstrates how to subscribe to stream $all without a filter.

We create two streams and write events to them. Subscription to stream $all must catch all events written to those two streams.

package main

import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/core/systemmetadata"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, stdErr := connection.ParseConnectionString(clientURI)
	if stdErr != nil {
		log.Fatalln(stdErr)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	createEvents := func(count uint32) event_streams.ProposedEventList {
		result := make(event_streams.ProposedEventList, 10)
		for i := uint32(0); i < count; i++ {
			result[i] = event_streams.ProposedEvent{
				EventId:      uuid.Must(uuid.NewRandom()),
				EventType:    "TestEvent",
				ContentType:  "application/octet-stream",
				UserMetadata: []byte{},
				Data:         []byte("some event data"),
			}
		}
		return result
	}

	wg := sync.WaitGroup{}
	wg.Add(1)

	firstStream := "firstStream"
	secondStream := "secondStream"

	beforeEvents := createEvents(10)
	afterEvents := createEvents(10)

	allUserEvents := append(beforeEvents, afterEvents...)

	// create a stream with some events in it
	_, err := client.AppendToStream(context.Background(),
		firstStream,
		stream_revision.WriteStreamRevisionNoStream{},
		beforeEvents)
	if err != nil {
		log.Fatalln(err)
	}

	// create a subscription to a stream
	streamReader, err := client.SubscribeToStreamAll(context.Background(),
		stream_revision.ReadPositionAllStart{},
		false)
	if err != nil {
		log.Fatalln(err)
	}

	go func() {
		defer wg.Done()

		var resultsRead event_streams.ProposedEventList
		readResult, err := streamReader.ReadOne()
		if err != nil {
			log.Fatalln(err)
		}

		if event, isEvent := readResult.GetEvent(); isEvent {
			if !systemmetadata.IsSystemStream(event.Event.StreamId) {
				resultsRead = append(resultsRead, event.ToProposedEvent())
			}
		}

		// if we have read all user defined event stop listening for events and return from go routine
		if reflect.DeepEqual(allUserEvents, resultsRead) {
			streamReader.Close()
			return
		}
	}()

	// append some events to a stream after a listening go routine has started
	_, err = client.AppendToStream(context.Background(),
		secondStream,
		stream_revision.WriteStreamRevisionNoStream{},
		afterEvents)
	if err != nil {
		log.Fatalln(err)
	}

	// wait for subscription to receive all events
	wg.Wait()
}

Output:

Example of trying to put a tombstone on a stream which does not exist.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamName := "some_stream"

	// tombstone a non-existing stream
	tombstoneResult, err := client.TombstoneStream(context.Background(),
		streamName,
		stream_revision.WriteStreamRevisionNoStream{})
	tombstonePosition, isPosition := tombstoneResult.GetPosition()

	// result of a hard-delete must be a position
	if !isPosition {
		log.Fatalln("Must be a position")
	}

	// position returned by hard-delete must not be zero
	if tombstonePosition.CommitPosition == 0 || tombstonePosition.PreparePosition == 0 {
		log.Fatalln("Commit and Prepare position must not be zero")
	}
}

Output:

Example of putting a tombstone on an existing stream.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/google/uuid"
	"github.com/pivonroll/EventStore-Client-Go/core/connection"
	"github.com/pivonroll/EventStore-Client-Go/core/stream_revision"
	"github.com/pivonroll/EventStore-Client-Go/event_streams"
)

func main() {
	username := "admin"
	password := "changeit"
	eventStoreEndpoint := "localhost:2113" // assuming that EventStoreDB is running on port 2113
	clientURI := fmt.Sprintf("esdb://%s:%s@%s", username, password, eventStoreEndpoint)
	config, err := connection.ParseConnectionString(clientURI)
	if err != nil {
		log.Fatalln(err)
	}
	grpcClient := connection.NewGrpcClient(*config)
	client := event_streams.NewClient(grpcClient)

	streamName := "some_stream"
	proposedEvent := event_streams.ProposedEvent{
		EventId:      uuid.Must(uuid.NewRandom()),
		EventType:    "TestEvent",
		ContentType:  "application/octet-stream",
		UserMetadata: []byte{},
		Data:         []byte("some event data"),
	}

	// create a stream with one event
	writeResult, err := client.AppendToStream(context.Background(),
		streamName,
		stream_revision.WriteStreamRevisionNoStream{},
		[]event_streams.ProposedEvent{proposedEvent})
	if err != nil {
		log.Fatalln(err)
	}

	// put tombstone on a stream
	tombstoneResult, err := client.TombstoneStream(context.Background(),
		streamName,
		stream_revision.WriteStreamRevision{Revision: writeResult.GetCurrentRevision()})

	tombstonePosition, isPosition := tombstoneResult.GetPosition()

	if !isPosition {
		log.Fatalln("Must be a position")
	}

	writePosition, _ := writeResult.GetPosition()

	// position returned after we put a tombstone on a stream must be greater
	// than the one of the last event
	if !tombstonePosition.GreaterThan(writePosition) {
		log.Fatalln("Delete position must be greater than last event's write position")
	}
}

Output: