Skip to content

Commit 3930d99

Browse files
halgarihalgari
committed
InMemory tests are working
1 parent be83c4a commit 3930d99

File tree

4 files changed

+46
-87
lines changed

4 files changed

+46
-87
lines changed

‎src/NexusMods.EventSourcing.RocksDB/RocksDBEventStore.cs

Lines changed: 5 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -40,72 +40,15 @@ public RocksDBEventStore(TSerializer serializer, Settings settings, ISerializati
4040
_deserializer = new SpanDeserializer<TSerializer>(serializer);
4141
}
4242

43-
44-
public override TransactionId Add<T>(T eventValue)
43+
public override TransactionId Add<T>(T eventEntity, (IIndexableAttribute, IAccumulator)[] indexed)
4544
{
46-
lock (this)
47-
{
48-
_tx = _tx.Next();
49-
50-
// Write the event itself
51-
{
52-
Span<byte> keySpan = stackalloc byte[8];
53-
_tx.WriteTo(keySpan);
54-
var span = _serializer.Serialize(eventValue);
55-
_db.Put(keySpan, span, _eventsColumn);
56-
}
57-
58-
// Update the entity indexes to mark them as having this event
59-
{
60-
var ingester = new ModifiedEntitiesIngester();
61-
eventValue.Apply(ingester);
62-
Span<byte> keySpan = stackalloc byte[24];
63-
_tx.WriteTo(keySpan.SliceFast(16..));
64-
foreach (var entityId in ingester.Entities)
65-
{
66-
entityId.TryWriteBytes(keySpan);
67-
_db.Put(keySpan, keySpan, _entityIndexColumn);
68-
}
69-
}
70-
return _tx;
71-
}
45+
throw new NotImplementedException();
7246
}
7347

74-
public override void EventsForEntity<TIngester>(EntityId entityId, TIngester ingester, TransactionId fromId, TransactionId toId)
48+
public override void EventsForIndex<TIngester, TVal>(IIndexableAttribute<TVal> attr, TVal value, TIngester ingester, TransactionId fromTx,
49+
TransactionId toTx)
7550
{
76-
Span<byte> startKey = stackalloc byte[24];
77-
entityId.TryWriteBytes(startKey);
78-
BinaryPrimitives.WriteUInt64BigEndian(startKey.SliceFast(16), fromId.Value);
79-
Span<byte> endKey = stackalloc byte[24];
80-
entityId.TryWriteBytes(endKey);
81-
if (toId == TransactionId.Max)
82-
BinaryPrimitives.WriteUInt64BigEndian(endKey.SliceFast(16), ulong.MaxValue);
83-
else
84-
BinaryPrimitives.WriteUInt64BigEndian(endKey.SliceFast(16), toId.Value + 1);
85-
86-
var options = new ReadOptions();
87-
unsafe
88-
{
89-
fixed (byte* startKeyPtr = startKey)
90-
{
91-
fixed (byte* endKeyPtr = endKey)
92-
{
93-
options.SetIterateUpperBound(endKeyPtr, 24);
94-
options.SetIterateLowerBound(startKeyPtr, 24);
95-
using var iterator = _db.NewIterator(_entityIndexColumn, options);
96-
97-
iterator.SeekToFirst();
98-
while (iterator.Valid())
99-
{
100-
var key = iterator.GetKeySpan().SliceFast(16);
101-
var txId = TransactionId.From(key);
102-
var evt = _db.Get(key, _deserializer, _eventsColumn);
103-
if (!ingester.Ingest(txId, evt)) break;
104-
iterator.Next();
105-
}
106-
}
107-
}
108-
}
51+
throw new NotImplementedException();
10952
}
11053

11154
public override TransactionId GetSnapshot(TransactionId asOf, EntityId entityId, out IAccumulator loadedDefinition,

‎src/NexusMods.EventSourcing/AEventStore.cs

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -98,10 +98,6 @@ protected ReadOnlySpan<byte> SerializeSnapshot(EntityId id, IDictionary<IAttribu
9898
var span = _writer.GetWrittenSpan();
9999
return span;
100100
}
101-
102-
public abstract TransactionId Add<T>(T eventEntity) where T : IEvent;
103-
public abstract void EventsForEntity<TIngester>(EntityId entityId, TIngester ingester, TransactionId fromId, TransactionId toId) where TIngester : IEventIngester;
104-
105101
public abstract TransactionId Add<T>(T eventEntity, (IIndexableAttribute, IAccumulator)[] indexed) where T : IEvent;
106102

107103
public abstract void EventsForIndex<TIngester, TVal>(IIndexableAttribute<TVal> attr, TVal value, TIngester ingester, TransactionId fromTx,

‎tests/NexusMods.EventSourcing.TestModel/InMemoryEventStore.cs

Lines changed: 40 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,16 @@
22
using System.Buffers.Binary;
33
using NexusMods.EventSourcing.Abstractions;
44
using NexusMods.EventSourcing.Abstractions.Serialization;
5+
using NexusMods.Hashing.xxHash64;
56
using Reloaded.Memory.Extensions;
67

78
namespace NexusMods.EventSourcing.TestModel;
89

910
public class InMemoryEventStore<TSerializer> : AEventStore
1011
where TSerializer : IEventSerializer
1112
{
12-
private TransactionId _tx = TransactionId.From(0);
13-
private readonly Dictionary<EntityId,IList<(TransactionId TxId, byte[] Data)>> _events = new();
13+
private readonly List<byte[]> _events = new();
14+
private readonly Dictionary<(IAttribute, Hash), SortedSet<TransactionId>> _indexes = new();
1415
private readonly Dictionary<EntityId, SortedDictionary<TransactionId, byte[]>> _snapshots = new();
1516
private TSerializer _serializer;
1617

@@ -19,43 +20,61 @@ public InMemoryEventStore(TSerializer serializer, ISerializationRegistry seriali
1920
_serializer = serializer;
2021
}
2122

22-
public override TransactionId Add<T>(T entity)
23+
public override TransactionId Add<T>(T entity, (IIndexableAttribute, IAccumulator)[] indexed)
2324
{
2425
lock (this)
2526
{
26-
_tx = _tx.Next();
27+
// Create the new txId
28+
var txId = TransactionId.From((ulong)_events.Count);
29+
2730
var data = _serializer.Serialize(entity);
28-
var logger = new ModifiedEntitiesIngester();
29-
entity.Apply(logger);
30-
foreach (var id in logger.Entities)
31+
_events.Add(data.ToArray());
32+
33+
foreach (var (attr, accumulator) in indexed)
3134
{
32-
if (!_events.TryGetValue(id, out var value))
35+
// Hash the accumulator to condense it down into a single ulong
36+
var hash = HashAccumulator(attr, accumulator);
37+
38+
if (_indexes.TryGetValue((attr, hash), out var found))
3339
{
34-
value = new List<(TransactionId, byte[])>();
35-
_events.Add(id, value);
40+
found.Add(txId);
41+
}
42+
else
43+
{
44+
var newSet = new SortedSet<TransactionId> { txId };
45+
_indexes.Add((attr, hash), newSet);
3646
}
37-
38-
value.Add((_tx, data.ToArray()));
3947
}
40-
41-
return _tx;
48+
return txId;
4249
}
4350
}
4451

52+
private static Hash HashAccumulator(IIndexableAttribute attr, IAccumulator accumulator)
53+
{
54+
Span<byte> span = stackalloc byte[attr.SpanSize()];
55+
attr.WriteTo(span, accumulator);
56+
var hash = span.XxHash64();
57+
return hash;
58+
}
4559

46-
public override void EventsForEntity<TIngester>(EntityId entityId, TIngester ingester, TransactionId fromId, TransactionId toId)
60+
public override void EventsForIndex<TIngester, TVal>(IIndexableAttribute<TVal> attr, TVal value, TIngester ingester, TransactionId fromTx,
61+
TransactionId toTx)
4762
{
48-
if (!_events.TryGetValue(entityId, out var events))
63+
Span<byte> valueSpan = stackalloc byte[attr.SpanSize()];
64+
attr.WriteTo(valueSpan, value);
65+
var hash = valueSpan.XxHash64();
66+
67+
if (!_indexes.TryGetValue((attr, hash), out var found))
4968
return;
5069

51-
foreach (var data in events)
70+
foreach (var txId in found)
5271
{
53-
if (data.TxId < fromId) continue;
54-
if (data.TxId > toId) break;
72+
if (txId > toTx || txId < fromTx) continue;
5573

56-
var @event = _serializer.Deserialize(data.Data)!;
57-
if (!ingester.Ingest(data.TxId, @event)) break;
74+
var eventItem = _serializer.Deserialize(_events[(int)txId.Value]);
75+
ingester.Ingest(txId, eventItem);
5876
}
77+
5978
}
6079

6180
public override TransactionId GetSnapshot(TransactionId asOf, EntityId entityId, out IAccumulator loadedDefinition,

‎tests/NexusMods.EventSourcing.TestModel/NexusMods.EventSourcing.TestModel.csproj

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
<ItemGroup>
1515
<PackageReference Include="MemoryPack" Version="1.10.0" />
16+
<PackageReference Include="NexusMods.Hashing.xxHash64" Version="1.0.1" />
1617
</ItemGroup>
1718

1819
</Project>

0 commit comments

Comments
 (0)