Querying Event and Stream Data #

public void load_event_stream(IDocumentSession session, Guid streamId)
{
    // Fetch *all* of the events for this stream
    var events1 = session.Events.FetchStream(streamId);

    // Fetch the events for this stream up to and including version 5
    var events2 = session.Events.FetchStream(streamId, 5);

    // Fetch the events for this stream at this time yesterday
    var events3 = session.Events
                .FetchStream(streamId, timestamp: DateTime.UtcNow.AddDays(-1));
}

public async Task load_event_stream_async(IDocumentSession session, Guid streamId)
{
    // Fetch *all* of the events for this stream
    var events1 = await session.Events.FetchStreamAsync(streamId);

    // Fetch the events for this stream up to and including version 5
    var events2 = await session.Events.FetchStreamAsync(streamId, 5);

    // Fetch the events for this stream at this time yesterday
    var events3 = await session.Events
                .FetchStreamAsync(streamId, timestamp: DateTime.UtcNow.AddDays(-1));
}

public interface IEvent
{
    /// <summary>
    /// Unique identifier for the event. Uses a sequential Guid
    /// </summary>
    Guid Id { get; set; }

    /// <summary>
    /// The version of the stream this event reflects. The place in the stream.
    /// </summary>
    long Version { get; set; }

    /// <summary>
    /// The sequential order of this event in the entire event store
    /// </summary>
    long Sequence { get; set; }

    /// <summary>
    ///     The actual event data body
    /// </summary>
    object Data { get; }

    /// <summary>
    ///     If using Guid's for the stream identity, this will
    ///     refer to the Stream's Id, otherwise it will always be Guid.Empty
    /// </summary>
    Guid StreamId { get; set; }

    /// <summary>
    ///     If using strings as the stream identifier, this will refer
    ///     to the containing Stream's Id
    /// </summary>
    string? StreamKey { get; set; }

    /// <summary>
    ///     The UTC time that this event was originally captured
    /// </summary>
    DateTimeOffset Timestamp { get; set; }

    /// <summary>
    ///     If using multi-tenancy by tenant id
    /// </summary>
    string TenantId { get; set; }

    /// <summary>
    /// The .Net type of the event body
    /// </summary>
    Type EventType { get; }

    /// <summary>
    /// Marten's type alias string for the Event type
    /// </summary>
    string EventTypeName { get; set; }

    /// <summary>
    /// Marten's string representation of the event type
    /// in assembly qualified name
    /// </summary>
    string DotNetTypeName { get; set; }

    /// <summary>
    /// Optional metadata describing the causation id
    /// </summary>
    string? CausationId { get; set; }

    /// <summary>
    /// Optional metadata describing the correlation id
    /// </summary>
    string? CorrelationId { get; set; }

    /// <summary>
    /// Optional user defined metadata values. This may be null.
    /// </summary>
    Dictionary<string, object>? Headers { get; set; }

    /// <summary>
    /// Set an optional user defined metadata value by key
    /// </summary>
    /// <param name="key"></param>
    /// <param name="value"></param>
    void SetHeader(string key, object value);

    /// <summary>
    /// Get an optional user defined metadata value by key
    /// </summary>
    /// <param name="key"></param>
    /// <returns></returns>
    object? GetHeader(string key);

    /// <summary>
    /// Has this event been archived and no longer applicable
    /// to projected views
    /// </summary>
    bool IsArchived { get; set; }
}

public class fetching_stream_state: IntegrationContext
{
    private Guid theStreamId;

    public fetching_stream_state(DefaultStoreFixture fixture) : base(fixture)
    {
        var joined = new MembersJoined { Members = new string[] { "Rand", "Matt", "Perrin", "Thom" } };
        var departed = new MembersDeparted { Members = new[] { "Thom" } };

        theStreamId = theSession.Events.StartStream<Quest>(joined, departed).Id;
        theSession.SaveChanges();
    }

    [Fact]
    public void can_fetch_the_stream_version_and_aggregate_type()
    {
        var state = theSession.Events.FetchStreamState(theStreamId);

        state.Id.ShouldBe(theStreamId);
        state.Version.ShouldBe(2);
        state.AggregateType.ShouldBe(typeof(Quest));
        state.LastTimestamp.ShouldNotBe(DateTime.MinValue);
        state.Created.ShouldNotBe(DateTime.MinValue);
    }

    [Fact]
    public async Task can_fetch_the_stream_version_and_aggregate_type_async()
    {
        var state = await theSession.Events.FetchStreamStateAsync(theStreamId);

        state.Id.ShouldBe(theStreamId);
        state.Version.ShouldBe(2);
        state.AggregateType.ShouldBe(typeof(Quest));
        state.LastTimestamp.ShouldNotBe(DateTime.MinValue);
        state.Created.ShouldNotBe(DateTime.MinValue);
    }

    [Fact]
    public async Task can_fetch_the_stream_version_through_batch_query()
    {
        var batch = theSession.CreateBatchQuery();

        var stateTask = batch.Events.FetchStreamState(theStreamId);

        await batch.Execute();

        var state = await stateTask;

        state.Id.ShouldBe(theStreamId);
        state.Version.ShouldBe(2);
        state.AggregateType.ShouldBe(typeof(Quest));
        state.LastTimestamp.ShouldNotBe(DateTime.MinValue);
    }

    [Fact]
    public async Task can_fetch_the_stream_events_through_batch_query()
    {
        var batch = theSession.CreateBatchQuery();

        var eventsTask = batch.Events.FetchStream(theStreamId);

        await batch.Execute();

        var events = await eventsTask;

        events.Count.ShouldBe(2);
    }
}

public void load_a_single_event_synchronously(IDocumentSession session, Guid eventId)
{
    // If you know what the event type is already
    var event1 = session.Events.Load<MembersJoined>(eventId);

    // If you do not know what the event type is
    var event2 = session.Events.Load(eventId);
}

public async Task load_a_single_event_asynchronously(IDocumentSession session, Guid eventId)
{
    // If you know what the event type is already
    var event1 = await session.Events.LoadAsync<MembersJoined>(eventId);

    // If you do not know what the event type is
    var event2 = await session.Events.LoadAsync(eventId);
}

[Fact]
public void can_query_against_event_type()
{
    theSession.Events.StartStream<Quest>(joined1, departed1);
    theSession.Events.StartStream<Quest>(joined2, departed2);

    theSession.SaveChanges();

    theSession.Events.QueryRawEventDataOnly<MembersJoined>().Count().ShouldBe(2);
    theSession.Events.QueryRawEventDataOnly<MembersJoined>().ToArray().SelectMany(x => x.Members).Distinct()
        .OrderBy(x => x)
        .ShouldHaveTheSameElementsAs("Egwene", "Matt", "Nynaeve", "Perrin", "Rand", "Thom");

    theSession.Events.QueryRawEventDataOnly<MembersDeparted>()
        .Single(x => x.Members.Contains("Matt")).Id.ShouldBe(departed2.Id);
}

public void example_of_querying_for_event_data(IDocumentSession session, Guid stream)
{
    var events = session.Events.QueryAllRawEvents()
        .Where(x => x.StreamId == stream)
        .OrderBy(x => x.Sequence)
        .ToList();
}