Reading events

There are two options for reading events from EventStoreDB. You can either read from an individual stream or you can read from the $all stream. The later will return all events in the store.

Each event in EventStoreDB belongs to an individual stream name. When reading events you can pick the name of the stream you want to read from. Then you can choose whether to read that stream forwards or backwards.

All events have a StreamPosition, which is the place of the event in the stream, represented by a big int (unsigned 64-bit integer) and a Position that is the events logical position that is represented by CommitPosition and a PreparePosition. This means that when reading events you have to supply a different "position" depending on if you are reading from a stream or the $all stream.

TIP

Check connecting to EventStoreDB instructions to learn how to configure and use the client SDK.

Reading from a stream

You can read events from individual streams, both all the events, or just a few events from the stream. You can start reading from any position in the stream, and read events both forwards and backwards. It's only possible to read events from one stream at a time. You can also read events from the global event log, which spans across streams. Read more about it in the Read from $all section below.

Reading forwards

The simplest way to read a stream forwards is to supply a stream name, direction and revision to start from. This can either be a stream position Start or a big int (unsigned 64-bit integer):

events = client.get_stream(
    stream_name=stream_name,
    stream_position=0,
    limit=100,
)

const events = client.readStream("some-stream", {
  direction: FORWARDS,
  fromRevision: START,
  maxCount: 10,
});

const events = client.readStream<SomeEvent>("some-stream", {
  direction: FORWARDS,
  fromRevision: START,
  maxCount: 10,
});

ReadStreamOptions options = ReadStreamOptions.get()
        .forwards()
        .fromStart();

ReadResult result = client.readStream("some-stream", options)
        .get();

var events = client.ReadStreamAsync(
    Direction.Forwards,
    "some-stream",
    StreamPosition.Start
);

options := esdb.ReadStreamOptions{
    From:      esdb.Start{},
    Direction: esdb.Forwards,
}
stream, err := db.ReadStream(context.Background(), "some-stream", options, 100)

if err != nil {
    panic(err)
}

defer stream.Close()

let options = ReadStreamOptions::default()
    .position(StreamPosition::Start)
    .forwards();
let mut stream = client.read_stream("some-stream", &options).await?;

This will return an enumerable that can be iterated on:

for event in events:
    # Doing something productive with the event
    print(f"Event: {event}")

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

for (ResolvedEvent resolvedEvent : result.getEvents()) {
    RecordedEvent recordedEvent = resolvedEvent.getOriginalEvent();
    System.out.println(new ObjectMapper().writeValueAsString(recordedEvent.getEventData()));
}

await foreach (var @event in events) Console.WriteLine(Encoding.UTF8.GetString(@event.Event.Data.ToArray()));

for {
    event, err := stream.Recv()

    if errors.Is(err, io.EOF) {
        break
    }

    if err != nil {
        panic(err)
    }

    fmt.Printf("Event> %v", event)
}

while let Some(event) = stream.next().await? {
    let test_event = event.get_original_event().as_json::<TestEvent>()?;

    println!("Event> {:?}", test_event);
}

There are a number of additional arguments you can provide when reading a stream, listed below.

maxCount

Passing in the max count allows you to limit the number of events returned.

resolveLinkTos

When using projections to create new events you can set whether the generated events are pointers to existing events. Setting this value to true tells EventStoreDB to return the event as well as the event linking to it.

configureOperationOptions

You can use the configureOperationOptions argument to provide a function that will customise settings for each operation.

userCredentials

The userCredentials argument is optional. You can use it to override the default credentials specified when creating the client instance.

credentials = client.construct_call_credentials(
    username="admin",
    password="changeit",
)

events = client.get_stream(
    stream_name=stream_name,
    credentials=credentials,
)

const credentials = {
  username: "admin",
  password: "changeit",
};

const events = client.readStream("some-stream", {
  direction: FORWARDS,
  fromRevision: START,
  credentials,
  maxCount: 10,
});

const credentials = {
  username: "admin",
  password: "changeit",
};

const events = client.readStream<SomeEvent>("some-stream", {
  direction: FORWARDS,
  fromRevision: START,
  credentials,
  maxCount: 10,
});

ReadStreamOptions options = ReadStreamOptions.get()
        .forwards()
        .fromStart()
        .authenticated("admin", "changeit");

ReadResult result = client.readStream("some-stream", options)
        .get();

var result = client.ReadStreamAsync(
    Direction.Forwards,
    "some-stream",
    StreamPosition.Start,
    userCredentials: new UserCredentials("admin", "changeit"),
    cancellationToken: cancellationToken
);

options := esdb.ReadStreamOptions{
    From: esdb.Start{},
    Authenticated: &esdb.Credentials{
        Login:    "admin",
        Password: "changeit",
    },
}
stream, err := db.ReadStream(context.Background(), "some-stream", options, 100)

let options = ReadStreamOptions::default()
    .position(StreamPosition::Start)
    .authenticated(Credentials::new("admin", "changeit"));

let stream = client.read_stream("some-stream", &options).await;

Reading from a revision

Instead of providing the StreamPosition you can also provide a specific stream revision as a big int (unsigned 64-bit integer).

events = client.get_stream(
    stream_name=stream_name,
    stream_position=10,
)

const events = client.readStream("some-stream", {
  direction: FORWARDS,
  fromRevision: BigInt(10),
  maxCount: 20,
});

const events = client.readStream<SomeEvent>("some-stream", {
  direction: FORWARDS,
  fromRevision: BigInt(10),
  maxCount: 20,
});

ReadStreamOptions options = ReadStreamOptions.get()
        .forwards()
        .fromRevision(10)
        .maxCount(20);

ReadResult result = client.readStream("some-stream", options)
        .get();

var events = client.ReadStreamAsync(
    Direction.Forwards,
    "some-stream",
    10,
    20
);

ropts := esdb.ReadStreamOptions{
    From: esdb.Revision(10),
}

stream, err := db.ReadStream(context.Background(), "some-stream", ropts, 20)

if err != nil {
    panic(err)
}

defer stream.Close()

let options = ReadStreamOptions::default()
    .position(StreamPosition::Position(10))
    .max_count(20);
let mut stream = client.read_stream("some-stream", &options).await?;

Reading backwards

As well as being able to read a stream forwards you can also go backwards. When reading backwards, you need to set the stream position the end if you want to read all the events:

events = client.get_stream(
    stream_name=stream_name,
    backwards=True,
)

const events = client.readStream("some-stream", {
  direction: BACKWARDS,
  fromRevision: END,
  maxCount: 10,
});

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

const events = client.readStream<SomeEvent>("some-stream", {
  direction: BACKWARDS,
  fromRevision: END,
  maxCount: 10,
});

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

ReadStreamOptions options = ReadStreamOptions.get()
        .backwards()
        .fromEnd();

ReadResult result = client.readStream("some-stream", options)
        .get();

for (ResolvedEvent resolvedEvent : result.getEvents()) {
    RecordedEvent recordedEvent = resolvedEvent.getOriginalEvent();
    System.out.println(new ObjectMapper().writeValueAsString(recordedEvent.getEventData()));
}

var events = client.ReadStreamAsync(
    Direction.Backwards,
    "some-stream",
    StreamPosition.End
);

await foreach (var e in events) Console.WriteLine(Encoding.UTF8.GetString(e.Event.Data.ToArray()));

ropts := esdb.ReadStreamOptions{
    Direction: esdb.Backwards,
    From:      esdb.End{},
}

stream, err := db.ReadStream(context.Background(), "some-stream", ropts, 10)

if err != nil {
    panic(err)
}

defer stream.Close()

for {
    event, err := stream.Recv()

    if errors.Is(err, io.EOF) {
        break
    }

    if err != nil {
        panic(err)
    }

    fmt.Printf("Event> %v", event)
}

let options = ReadStreamOptions::default()
    .position(StreamPosition::End)
    .backwards();
let mut stream = client.read_stream("some-stream", &options).await?;

while let Some(event) = stream.next().await? {
    let test_event = event.get_original_event().as_json::<TestEvent>()?;

    println!("Event> {:?}", test_event);
}

TIP

You can use reading backwards to find the last position in the stream. Just read backwards one event and get the position.

Checking if the stream exists

Reading a stream returns a ReadStreamResult, which contains a property ReadState. This property can have the value StreamNotFound and Ok.

It is important to check the value of this field before attempting to iterate an empty stream as it will throw an exception.

For example:

try:
    events = client.get_stream(
        stream_name=unknown_stream_name, limit=1
    )
except exceptions.NotFound:
    print("The stream was not found")

const events = client.readStream("some-stream", {
  direction: FORWARDS,
  fromRevision: BigInt(10),
  maxCount: 20,
});

try {
  for await (const resolvedEvent of events) {
    console.log(resolvedEvent.event?.data);
  }
} catch (error) {
  if (error instanceof StreamNotFoundError) {
    return;
  }

  throw error;
}

const events = client.readStream<SomeEvent>("some-stream", {
  direction: FORWARDS,
  fromRevision: BigInt(10),
  maxCount: 20,
});

try {
  for await (const resolvedEvent of events) {
    console.log(resolvedEvent.event?.data);
  }
} catch (error) {
  if (error instanceof StreamNotFoundError) {
    return;
  }

  throw error;
}

ReadStreamOptions options = ReadStreamOptions.get()
        .forwards()
        .fromRevision(10)
        .maxCount(20);

ReadResult result = null;
try {
    result = client.readStream("some-stream", options)
            .get();
} catch (ExecutionException e) {
    Throwable innerException = e.getCause();

    if (innerException instanceof StreamNotFoundException) {
        return;
    }
}

for (ResolvedEvent resolvedEvent : result.getEvents()) {
    RecordedEvent recordedEvent = resolvedEvent.getOriginalEvent();
    System.out.println(new ObjectMapper().writeValueAsString(recordedEvent.getEventData()));
}

var result = client.ReadStreamAsync(
    Direction.Forwards,
    "some-stream",
    10,
    20
);

if (await result.ReadState == ReadState.StreamNotFound) return;

await foreach (var e in result) Console.WriteLine(Encoding.UTF8.GetString(e.Event.Data.ToArray()));

ropts := esdb.ReadStreamOptions{
    From: esdb.Revision(10),
}

stream, err := db.ReadStream(context.Background(), "some-stream", ropts, 100)

if err != nil {
    panic(err)
}

defer stream.Close()

for {
    event, err := stream.Recv()

    if err, ok := esdb.FromError(err); !ok {
        if err.Code() == esdb.ErrorCodeResourceNotFound {
            fmt.Print("Stream not found")
        } else if errors.Is(err, io.EOF) {
            break
        } else {
            panic(err)
        }
    }

    fmt.Printf("Event> %v", event)
}

let options = ReadStreamOptions::default().position(StreamPosition::Position(10));

let mut stream = client.read_stream("some-stream", &options).await?;

while let Some(event) = stream.next().await? {
    let test_event = event.get_original_event().as_json::<TestEvent>()?;

    println!("Event> {:?}", test_event);
}

Reading from the $all stream

Reading from the all stream is similar to reading from an individual stream but with some small differences. Primarily the need to provide an admin user account credentials and that you need to provide a transaction log position instead of a stream revision.

Reading forwards

The simplest way to read the $all stream forwards is to supply a direction and transaction log position to start from. This can either be a stream position Start or a big int (unsigned 64-bit integer):

events = client.read_all(
    commit_position=0,
    limit=100,
)

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  maxCount: 10,
});

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  maxCount: 10,
});

ReadAllOptions options = ReadAllOptions.get()
        .forwards()
        .fromStart();

ReadResult result = client.readAll(options)
        .get();

var events = client.ReadAllAsync(Direction.Forwards, Position.Start);

options := esdb.ReadAllOptions{
    From:      esdb.Start{},
    Direction: esdb.Forwards,
}
stream, err := db.ReadAll(context.Background(), options, 100)

if err != nil {
    panic(err)
}

defer stream.Close()

let options = ReadAllOptions::default()
    .position(StreamPosition::Start)
    .forwards();
let mut stream = client.read_all(&Default::default()).await?;

You can iterate asynchronously through the result:

for event in events:
    print(f"Event: {event}")

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

for await (const resolvedEvent of events) {
  console.log(resolvedEvent.event?.data);
}

for (ResolvedEvent resolvedEvent : result.getEvents()) {
    RecordedEvent recordedEvent = resolvedEvent.getOriginalEvent();
    System.out.println(new ObjectMapper().writeValueAsString(recordedEvent.getEventData()));
}

await foreach (var e in events) Console.WriteLine(Encoding.UTF8.GetString(e.Event.Data.ToArray()));

for {
    event, err := stream.Recv()

    if errors.Is(err, io.EOF) {
        break
    }

    if err != nil {
        panic(err)
    }

    fmt.Printf("Event> %v", event)
}

while let Some(event) = stream.next().await? {
    println!("Event> {:?}", event.get_original_event());
}

There are a number of additional arguments you can provide when reading a stream.

maxCount

Passing in the max count allows you to limit the number of events that returned.

resolveLinkTos

When using projections to create new events you can set whether the generated events are pointers to existing events. Setting this value to true will tell EventStoreDB to return the event as well as the event linking to it.

events = client.read_all(
    commit_position=0,
    limit=100,
    resolve_links=True,
)

const events = client.readAll({
  direction: BACKWARDS,
  fromPosition: END,
  resolveLinkTos: true,
  maxCount: 10,
});

const events = client.readAll({
  direction: BACKWARDS,
  fromPosition: END,
  resolveLinkTos: true,
  maxCount: 10,
});

ReadAllOptions options = ReadAllOptions.get()
        .forwards()
        .fromStart()
        .resolveLinkTos();

ReadResult result = client.readAll(options)
        .get();

var result = client.ReadAllAsync(
    Direction.Forwards,
    Position.Start,
    resolveLinkTos: true,
    cancellationToken: cancellationToken
);

ropts := esdb.ReadAllOptions{
    ResolveLinkTos: true,
}

stream, err := db.ReadAll(context.Background(), ropts, 100)

let options = ReadAllOptions::default().resolve_link_tos();
client.read_all(&options).await?;

configureOperationOptions

This argument is generic setting class for all operations that can be set on all operations executed against EventStoreDB.

userCredentials

The credentials used to read the data can be supplied. to be used by the subscription as follows. This will override the default credentials set on the connection.

credentials = client.construct_call_credentials(
    username="admin",
    password="changeit",
)

events = client.read_all(
    commit_position=0,
    limit=100,
    credentials=credentials,
)

const credentials = {
  username: "admin",
  password: "changeit",
};

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  credentials,
  maxCount: 10,
});

const credentials = {
  username: "admin",
  password: "changeit",
};

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  credentials,
  maxCount: 10,
});

ReadAllOptions options = ReadAllOptions.get()
        .forwards()
        .fromStart()
        .authenticated("admin", "changeit");

ReadResult result = client.readAll(options)
        .get();

var result = client.ReadAllAsync(
    Direction.Forwards,
    Position.Start,
    userCredentials: new UserCredentials("admin", "changeit"),
    cancellationToken: cancellationToken
);

ropts := esdb.ReadAllOptions{
    From: esdb.Start{},
    Authenticated: &esdb.Credentials{
        Login:    "admin",
        Password: "changeit",
    },
}
stream, err := db.ReadAll(context.Background(), ropts, 100)

let options = ReadAllOptions::default()
    .authenticated(Credentials::new("admin", "changeit"))
    .position(StreamPosition::Position(Position {
        commit: 1_110,
        prepare: 1_110,
    }));
let stream = client.read_all(&options).await;

Reading backwards

As well as being able to read a stream forwards you can also go backwards. When reading backwards is the position will have to be set to the end if you want to read all the events:

events = client.read_all(
    backwards=True,
    limit=100,
)

const events = client.readAll({
  direction: BACKWARDS,
  fromPosition: END,
  maxCount: 10,
});

const events = client.readAll({
  direction: BACKWARDS,
  fromPosition: END,
  maxCount: 10,
});

ReadAllOptions options = ReadAllOptions.get()
        .backwards()
        .fromEnd();

ReadResult result = client.readAll(options)
        .get();

var events = client.ReadAllAsync(Direction.Backwards, Position.End);

ropts := esdb.ReadAllOptions{
    Direction: esdb.Backwards,
    From:      esdb.End{},
}

stream, err := db.ReadAll(context.Background(), ropts, 100)

if err != nil {
    panic(err)
}

defer stream.Close()

let options = ReadAllOptions::default().position(StreamPosition::End);

let mut stream = client.read_all(&options).await?;

TIP

You can use reading backwards to find the last position in the stream. Just read backwards one event and get the position.

Handling system events

When reading from the all stream EventStoreDB will also return system events. In most cases you won't need to care about these events and they should be ignored.

All system events begin with $ or $$ and can be easily ignored by checking the EventType property.

events = client.read_all(
    filter_exclude=[],  # system events are excluded by default
)

for event in events:
    if event.type.startswith("$"):
        continue

    print(f"Event: {event.type}")

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  maxCount: 10,
});

for await (const resolvedEvent of events) {
  if (resolvedEvent.event?.type.startsWith("$")) {
    continue;
  }

  console.log(resolvedEvent.event?.type);
}

const events = client.readAll({
  direction: FORWARDS,
  fromPosition: START,
  maxCount: 10,
});

for await (const resolvedEvent of events) {
  if (resolvedEvent.event?.type.startsWith("$")) {
    continue;
  }

  console.log(resolvedEvent.event?.type);
}

ReadAllOptions options = ReadAllOptions.get()
        .forwards()
        .fromStart();

ReadResult result = client.readAll(options)
        .get();

for (ResolvedEvent resolvedEvent : result.getEvents()) {
    RecordedEvent recordedEvent = resolvedEvent.getOriginalEvent();
    if (recordedEvent.getEventType().startsWith("$")) {
        continue;
    }
    System.out.println(new ObjectMapper().writeValueAsString(recordedEvent.getEventData()));
}

var events = client.ReadAllAsync(Direction.Forwards, Position.Start);

await foreach (var e in events) {
    if (e.Event.EventType.StartsWith("$")) continue;

    Console.WriteLine(Encoding.UTF8.GetString(e.Event.Data.ToArray()));
}

stream, err := db.ReadAll(context.Background(), esdb.ReadAllOptions{}, 100)

if err != nil {
    panic(err)
}

defer stream.Close()

for {
    event, err := stream.Recv()

    if errors.Is(err, io.EOF) {
        break
    }

    if err != nil {
        panic(err)
    }

    fmt.Printf("Event> %v", event)

    if strings.HasPrefix(event.OriginalEvent().EventType, "$") {
        continue
    }

    fmt.Printf("Event> %v", event)
}

let mut stream = client.read_all(&Default::default()).await?;

while let Some(event) = stream.next().await? {
    if event.get_original_event().event_type.starts_with("$") {
        continue;
    }

    println!("Event> {:?}", event.get_original_event());
}