So GES returns a checkpoint at the end of the stream, I'll add a test and implementation for the other providers
nCrunch is having issues building the CES tests project, runs ok in cmdline test runner
GES client -> when I read to the end of the stream it returns a checkpoint and IsEndOfStream = true
when I read again from the given checkpoint, I get no events, IsEndOfStream = false and I get a new Checkpoint
Checkpoint will change(increase) for other reasons
that is global across all streams
but IsEndOfStream = false when reading past the end of a stream...
but even if you get IsEndOfStream = false, when you iterate the result page that has zero items, what is the real problem
or rather, is there a real problem
the problem is that when i get to the end of the stream, I still need a Checkpoint to persist
GES does return a checkpoint
but when you read from that checkpoint
i would expect {no events, isEndOfStream = true, a checkpoint}
What I have been doing to assert compatibility is to write a test to see what happens
i.e. not really on docs
yeah i've added a test
so if the GES behaviour seems weird, should prob ping the GES docs/peoples
(I'm interested too btw).
The good news is that we've managed to move all our events into Cedar, and project them all using a CatchUpProjection, but now we're at the end of the stream and our checkpoint is null :)
so close, yet so far!
as does GES client tests
it doesn't test trying to read beyond the end of the stream
Give me about hr, got a meeting
TestFixtureSetUp failed in TestsInitFixture
Exception doesn't have a stacktrace
Exception doesn't have a stacktrace
i grabbed js1.dll from the release binaries :trollface:
btw I have another branch
append-other-scenarios that has some failing tests for other edge cases. Going to fix that up shortly.
[Test, Category("LongRunning")]
public void be_return_empty_slice_with_position_if_asked_to_read_from_the_last_position()
{
var all = new List<RecordedEvent>();
var position = Position.Start;
AllEventsSlice slice;
while (!(slice = _conn.ReadAllEventsForwardAsync(position, 1, false).Result).IsEndOfStream)
{
all.Add(slice.Events.Single().Event);
position = slice.NextPosition;
}
Assert.That(EventDataComparer.Equal(_testEvents, all.Skip(all.Count - _testEvents.Length).ToArray()));
slice = _conn.ReadAllEventsForwardAsync(position, 1, false).Result;
Assert.That(slice.IsEndOfStream);
Assert.That(slice.Events.Length == 0);
Assert.That(slice.NextPosition == slice.FromPosition);
}
yeah I merged in all the bits from
append-other-scenarios
i'm not actually sure what the behaviour should be ^ the above test is speculation
on a related note, how are you finding the test structure in the CES solution? I need to come up with a way of reusable tests across all implementations; best I could come up with.
it works well enough :trollface: tbh i haven't given it much thought as I'm focussing on getting-shit-done right now :)
also I'm kind of familiar with it as it's similar to NEventStore so nothing i haven't seen before
try
{
var checkpointValue = (await dbConnection.QueryAsync<string>("SELECT checkpoint FROM projections.checkpoint;")).FirstOrDefault();
var checkpoint = checkpointValue != null ? new Checkpoint(checkpointValue) : Checkpoint.Start;
var events = await eventStore.ReadAll(checkpoint, 512);
foreach (var streamEvent in events.StreamEvents)
{
//dispatch events
var eventType = Type.GetType(streamEvent.Type, true, true);
var body = JsonSerializer.DeserializeFromString(streamEvent.JsonData, eventType) as IDomainEvent;
var headers = JsonSerializer.DeserializeFromString<IDictionary>(streamEvent.JsonMetadata);
var envelope = EventEnvelope.Create(body);
envelope.CopyHeaders(headers);
messageBus.Publish(envelope);
}
var nextCheckpoint = events.IsEnd ? Checkpoint.End : events.NextCheckpoint;
await dbConnection.ExecuteAsync("UPDATE projections.checkpoint SET checkpoint = @checkpoint;", new { checkpoint = nextCheckpoint.ToString() });
tx.Commit();
}var nextCheckpoint = events.IsEnd ? Checkpoint.End : events.NextCheckpoint; <-- this is completely wrongbasically read all, persist checkpoint, go to sleep 500ms, read all from persisted checkpoint, etc
so if i get to the end of the stream, i should get a checkpoint back that will give me any events after that checkpoint
Not getting IsEnd = true, nor am I getting a null checkpoint. Hmm.
yeah
going to put a loop in, see if it ever ends.
[Fact]
public async Task When_read_past_end_of_all()
{
using(var fixture = GetFixture())
{
using(var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
bool isEnd = false;
int count = 0;
Checkpoint checkpoint = Checkpoint.Start;
while (!isEnd)
{
_testOutputHelper.WriteLine($"Loop {count}");
var streamEventsPage = await eventStore.ReadAll(checkpoint, 10);
_testOutputHelper.WriteLine($"FromCheckpoint = {streamEventsPage.FromCheckpoint}");
_testOutputHelper.WriteLine($"NextCheckpoint = {streamEventsPage.NextCheckpoint}");
_testOutputHelper.WriteLine($"IsEnd = {streamEventsPage.IsEnd}");
_testOutputHelper.WriteLine($"StreamEvents.Count = {streamEventsPage.StreamEvents.Count}");
_testOutputHelper.WriteLine("");
checkpoint = streamEventsPage.NextCheckpoint;
isEnd = streamEventsPage.IsEnd;
count++;
if(count > 100)
{
throw new Exception("omg wtf");
}
}
}
}
}
results in
Loop 0
FromCheckpoint = 0/0
NextCheckpoint = 1624/1624
IsEnd = False
StreamEvents.Count = 3
Loop 1
FromCheckpoint = 1624/1624
NextCheckpoint = 2362/2362
IsEnd = False
StreamEvents.Count = 0
Loop 2
FromCheckpoint = 2362/2362
NextCheckpoint = 2362/2362
IsEnd = True
StreamEvents.Count = 0
ok, tried to format that
but it does end
GES might hand you some empty chunks but eventually you get to IsEndOfStream = true and NextCheckpoint == FromCheckpoint
your test passes for GES and Postgres
so i basically had to add
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
to my test, because CES may hand you an empty page from the end of the stream..
yes, Loops 0 had a page count 3, but loops 1 & 2 has a count of zero before loop 3 being IsEnd = true
so yeah end up a couple extra reads but no extra events.
but.. is there a race whereby, as a reader, you'll never reach the end?
The subscription API will be better for this anyway.
right now i'm just building a .exe to build out the read model
then poll for new events
try
{
var checkpointValue = (await dbConnection.QueryAsync<string>("SELECT checkpoint FROM projections.checkpoint;")).FirstOrDefault();
var checkpoint = checkpointValue != null ? new Checkpoint(checkpointValue) : Checkpoint.Start;
var events = await eventStore.ReadAll(checkpoint, 512);
foreach (var streamEvent in events.StreamEvents)
{
//dispatch events
var eventType = Type.GetType(streamEvent.Type, true, true);
var body = JsonSerializer.DeserializeFromString(streamEvent.JsonData, eventType) as IDomainEvent;
var headers = JsonSerializer.DeserializeFromString<IDictionary>(streamEvent.JsonMetadata);
var envelope = EventEnvelope.Create(body);
envelope.CopyHeaders(headers);
messageBus.Publish(envelope);
}
await dbConnection.ExecuteAsync("UPDATE projections.checkpoint SET checkpoint = @checkpoint;", new { checkpoint = events.NextCheckpoint });
tx.Commit();
}
yes that's it
i was testing for isEnd to figure out what the checkpoint would be
that was a red herring
the original code looked like that
shall i add in the end-of-stream checkpoint for MSSQL/Sqlite or just focus on postgres in this branch and deal with that later?
This test is failing for MsSql, going to fix it now.
[Fact]
public async Task Read_forwards_to_the_end_should_return_a_valid_Checkpoint()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3, 4, 5, 6));
// read to the end of the stream
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, 4);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.IsEnd.Should().BeTrue();
Checkpoint currentCheckpoint = allEventsPage.NextCheckpoint;
currentCheckpoint.Should().NotBeNull();
// read end of stream again, should be empty, should return same checkpoint
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.StreamEvents.Should().BeEmpty();
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
currentCheckpoint = allEventsPage.NextCheckpoint;
// append some events then read again from the saved checkpoint, the next checkpoint should have moved
await eventStore.AppendToStream("stream-1", ExpectedVersion.Any, CreateNewStreamEvents(7, 8, 9));
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
allEventsPage.NextCheckpoint.Should().NotBe(currentCheckpoint.Value);
}
}
}
that'll go into infinite loop in a bad impl?
right, I'll co-opt this, but will add a short circuit for that while loop
tbh the while loop is copied from another test
ah wait you have a counter in there
i took it out to focus the assertions on the case in point
[Fact]
public async Task Read_forwards_to_the_end_should_return_a_valid_Checkpoint()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3, 4, 5, 6));
// read to the end of the stream
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, 4);
int count = 0; //counter is used to short circuit bad implementations that never return IsEnd = true
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.IsEnd.Should().BeTrue();
Checkpoint currentCheckpoint = allEventsPage.NextCheckpoint;
currentCheckpoint.Should().NotBeNull();
// read end of stream again, should be empty, should return same checkpoint
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
count = 0;
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.StreamEvents.Should().BeEmpty();
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
currentCheckpoint = allEventsPage.NextCheckpoint;
// append some events then read again from the saved checkpoint, the next checkpoint should have moved
await eventStore.AppendToStream("stream-1", ExpectedVersion.Any, CreateNewStreamEvents(7, 8, 9));
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
count = 0;
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
allEventsPage.NextCheckpoint.Should().NotBe(currentCheckpoint.Value);
}
}
}
I also want to console out some info.
_tesOutputHelper
prob split this into 2 or 3 tests, it's a bit on the large side.
replay 1.3million events , make a cup of coffee
namespace Cedar.EventStore
{
using System.Threading.Tasks;
using FluentAssertions;
using Xunit;
public abstract partial class EventStoreAcceptanceTests
{
[Fact]
public async Task Can_read_to_end_of_allstream()
{
using(var fixture = GetFixture())
{
using(var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
var lastAllEventPage = await ReadAllStreamToEnd(eventStore);
lastAllEventPage.StreamEvents.Should().BeEmpty();
lastAllEventPage.IsEnd.Should().BeTrue();
lastAllEventPage.NextCheckpoint.Should().Be(lastAllEventPage.FromCheckpoint);
}
}
}
[Fact]
public async Task When_read_to_end_of_allstream_then_append_event_then_can_read_to_end_of_allstream()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
await ReadAllStreamToEnd(eventStore);
await eventStore.AppendToStream("stream-1", 2, CreateNewStreamEvents(1, 2, 3));
var lastAllEventPage = await ReadAllStreamToEnd(eventStore);
lastAllEventPage.StreamEvents.Should().BeEmpty();
lastAllEventPage.IsEnd.Should().BeTrue();
lastAllEventPage.NextCheckpoint.Should().Be(lastAllEventPage.NextCheckpoint);
}
}
}
private async Task<AllEventsPage> ReadAllStreamToEnd(IEventStore eventStore)
{
int pageSize = 4, count = 0;
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, pageSize);
LogAllEventsPage(allEventsPage);
while (!allEventsPage.IsEnd && count < 10)
{
_testOutputHelper.WriteLine($"Loop {count}");
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, pageSize);
LogAllEventsPage(allEventsPage);
count++;
}
return allEventsPage;
}
private void LogAllEventsPage(AllEventsPage allEventsPage)
{
_testOutputHelper.WriteLine($"FromCheckpoint = {allEventsPage.FromCheckpoint}");
_testOutputHelper.WriteLine($"NextCheckpoint = {allEventsPage.NextCheckpoint}");
_testOutputHelper.WriteLine($"IsEnd = {allEventsPage.IsEnd}");
_testOutputHelper.WriteLine($"StreamEvents.Count = {allEventsPage.StreamEvents.Count}");
_testOutputHelper.WriteLine("");
}
}}
will merge it with append-other-scenarios
actually, I think I'll merge that branch in also
MSSql tests failing on master anyway.
Then I'll restart fixing them
ok on master
One failing GES test, will have a look aprés lunch
i changed
"data", "metadata" to "\"data\"", "\"metadata\"" since postgres checks for valid json on input
i've got When_read_to_end_of_allstream_then_append_then_can_read_to_end_of_allstream StreamEvents.Should().Be.Empty() failing
the failing test is because GES is treating the second append idempotently oop
actually, the idempotent area is one thing I don't know how I'm going to handle.
Added some idempotency tests. Passing on GES, failing on MSSql oc
i have cheated here
i used the NES commit id
so i could trace fuckups back to the source
i guess you could put a unique index on it and swallow the exception
basically if you give it an expected version and the event ids match whatever event Id are in that place, then all is ok
i.e.
A) Append ExpectedVersion= 0, A,B,C,D,E,F
then
B) Append ExpectedVersion= 0, A,B,C,D,E,F
A) Append ExpectedVersion= 0, A,B,C,D,E,F
then
B) Append ExpectedVersion= 0, A,B,C,D,E,F
it's the same op
so no exceptions
i guess we're following GES so just go with whatever they do eh?
also these will not throw either
C) Append ExpectedVersion = 1; B,C,D,E,F
D) Append ExpectedVersion= 3; D,E
C) Append ExpectedVersion = 1; B,C,D,E,F
D) Append ExpectedVersion= 3; D,E
Sure, I understand that, I'm just wondering what kind of situations this is useful in?
the fact that GES is not transactional
and is usable over HTTP I guess
nor DTC
I'm imagining an HTTP operation that succeeds server side but client doesn't get response
Client thenretries
am open to better names on those test btw
so you could knock up some OWIN bits to build yourself an EventStore knockoff on a RDBMS backend
note could not should heh
I might actually have to for our own reasons.
MSSql is borked.
I'll have to come back to this later. Lunch time!
oc, you could get postgres working on these test (everything on master). Would :heart: a review on them too.
apart from one
streamEvents.Empty on end-of stream
also nCrunch stubbornly refusing to load GetEventStore.Tests
updating EventStore.Client in GES.Tests to 3.0.2 (inline with Cedar.ES.GES) seemed to sort it out
ah
js1.dll not found
actually i suspect i might have commented that out to get it working too haha
(the post-build thingy)
that shit keeps happening to 2 projects and I don't know whay
why
GesEventStoreFixture.cs (18, 51): The type 'IEventStoreConnection' is defined in an assembly that is not referenced. You must add a reference to assembly 'EventStore.ClientAPI, Version=3.0.2.0, Culture=neutral, PublicKeyToken=null'.
cool
update-package eventstore.client
there's a surprise
right, lunch.