Skip to content

Commit 21101da

Browse files
nanovnanov
committed
add position in store to mongodb and inmemory implementations
1 parent cc70858 commit 21101da

3 files changed

Lines changed: 184 additions & 59 deletions

File tree

‎lib/databases/inmemory.js‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ function InMemory(options) {
1010
this.snapshots = {};
1111
this.undispatchedEvents = { _direct: {} };
1212
this.options = options;
13+
if (options.trackPosition)
14+
this.position = 0;
1315
}
1416

1517
util.inherits(InMemory, Store);
@@ -57,6 +59,7 @@ _.extend(InMemory.prototype, {
5759
this.store = {};
5860
this.snapshots = {};
5961
this.undispatchedEvents = { _direct: {} };
62+
this.position = 0;
6063
if (callback) callback(null);
6164
},
6265

@@ -93,6 +96,9 @@ _.extend(InMemory.prototype, {
9396

9497
var self = this;
9598
_.forEach(events, function(evt) {
99+
if (self.options.trackPosition) {
100+
evt.position = ++self.position;
101+
}
96102
self.undispatchedEvents._direct[evt.id] = evt;
97103
});
98104

‎lib/databases/mongodb.js‎

Lines changed: 84 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -134,8 +134,10 @@ _.extend(Mongo.prototype, {
134134
}
135135

136136
var client;
137+
var ensureIndex = 'ensureIndex';
137138

138139
if (mongo.MongoClient.length === 2) {
140+
ensureIndex = 'createIndex';
139141
client = new mongo.MongoClient(connectionUrl, options.options);
140142
client.connect(function(err, cl) {
141143
if (err) {
@@ -179,23 +181,28 @@ _.extend(Mongo.prototype, {
179181
}
180182

181183
self.events = self.db.collection(options.eventsCollectionName);
182-
self.events.ensureIndex({ aggregateId: 1, streamRevision: 1 },
184+
self.events[ensureIndex]({ aggregateId: 1, streamRevision: 1 },
183185
function (err) { if (err) { debug(err); } });
184-
self.events.ensureIndex({ commitStamp: 1 },
186+
self.events[ensureIndex]({ commitStamp: 1 },
185187
function (err) { if (err) { debug(err); } });
186-
self.events.ensureIndex({ dispatched: 1 }, { sparse: true },
188+
self.events[ensureIndex]({ dispatched: 1 }, { sparse: true },
187189
function (err) { if (err) { debug(err); } });
188-
self.events.ensureIndex({ commitStamp: 1, streamRevision: 1, commitSequence: 1 },
190+
self.events[ensureIndex]({ commitStamp: 1, streamRevision: 1, commitSequence: 1 },
189191
function (err) { if (err) { debug(err); } });
190192

191193
self.snapshots = self.db.collection(options.snapshotsCollectionName);
192-
self.snapshots.ensureIndex({ aggregateId: 1, revision: -1 },
194+
self.snapshots[ensureIndex]({ aggregateId: 1, revision: -1 },
193195
function (err) { if (err) { debug(err); } });
194196

195197
self.transactions = self.db.collection(options.transactionsCollectionName);
196-
self.transactions.ensureIndex({ aggregateId: 1, 'events.streamRevision': 1 },
198+
self.transactions[ensureIndex]({ aggregateId: 1, 'events.streamRevision': 1 },
197199
function (err) { if (err) { debug(err); } });
198200

201+
if (options.positionsCollectionName) {
202+
self.positions = self.db.collection(options.positionsCollectionName);
203+
self.positionsCounterId = options.eventsCollectionName;
204+
}
205+
199206
self.emit('connect');
200207
if (self.options.heartbeat) {
201208
self.startHeartbeat();
@@ -263,6 +270,11 @@ _.extend(Mongo.prototype, {
263270
},
264271
function (callback) {
265272
self.transactions.remove({}, callback);
273+
},
274+
function (callback) {
275+
if (!self.positions)
276+
return callback(null);
277+
self.positions.remove({}, callback);
266278
}
267279
], function (err) {
268280
if (err) {
@@ -276,6 +288,20 @@ _.extend(Mongo.prototype, {
276288
callback(null, new ObjectID().toString());
277289
},
278290

291+
getNextPositions: function(positions, callback) {
292+
if (!this.positions)
293+
return callback(null);
294+
295+
this.positions.findOneAndUpdate({ _id: this.positionsCounterId }, { $inc: { position: positions } }, { returnOriginal: false, upsert: true }, function (err, pos) {
296+
if (err)
297+
return callback(err);
298+
299+
pos.value.position += 1;
300+
301+
callback(null, _.range(pos.value.position - positions, pos.value.position));
302+
});
303+
},
304+
279305
addEvents: function (events, callback) {
280306
if (events.length === 0) {
281307
if (callback) { callback(null); }
@@ -287,64 +313,70 @@ _.extend(Mongo.prototype, {
287313
var noAggregateId = false,
288314
invalidCommitId = false;
289315

290-
_.forEach(events, function (evt) {
291-
if (!evt.aggregateId) {
292-
noAggregateId = true;
293-
}
294-
295-
if (!evt.commitId || evt.commitId !== commitId) {
296-
invalidCommitId = true;
316+
var self = this;
317+
this.getNextPositions(events.length, function(err, positions) {
318+
if (err) {
319+
debug(err);
320+
if (callback) callback(err);
321+
return;
297322
}
323+
_.forEach(events, function (evt, index) {
324+
if (!evt.aggregateId) {
325+
noAggregateId = true;
326+
}
298327

299-
evt._id = evt.id;
300-
evt.dispatched = false;
301-
});
302-
303-
if (noAggregateId) {
304-
var errMsg = 'aggregateId not defined!';
305-
debug(errMsg);
306-
if (callback) callback(new Error(errMsg));
307-
return;
308-
}
309-
310-
if (invalidCommitId) {
311-
var errMsg = 'commitId not defined or different!';
312-
debug(errMsg);
313-
if (callback) callback(new Error(errMsg));
314-
return;
315-
}
316-
317-
var self = this;
328+
if (!evt.commitId || evt.commitId !== commitId) {
329+
invalidCommitId = true;
330+
}
318331

319-
if (events.length === 1) {
320-
return this.events.insert(events, callback);
321-
}
332+
evt._id = evt.id;
333+
evt.dispatched = false;
334+
if (positions)
335+
evt.position = positions[index];
336+
});
322337

323-
var tx = {
324-
_id: commitId,
325-
events: events,
326-
aggregateId: events[0].aggregateId,
327-
aggregate: events[0].aggregate,
328-
context: events[0].context
329-
};
338+
if (noAggregateId) {
339+
var errMsg = 'aggregateId not defined!';
340+
debug(errMsg);
341+
if (callback) callback(new Error(errMsg));
342+
return;
343+
}
330344

331-
this.transactions.insert(tx, function (err) {
332-
if (err) {
333-
debug(err);
334-
if (callback) callback(err);
345+
if (invalidCommitId) {
346+
var errMsg = 'commitId not defined or different!';
347+
debug(errMsg);
348+
if (callback) callback(new Error(errMsg));
335349
return;
336350
}
337351

338-
self.events.insert(events, function (err) {
352+
if (events.length === 1) {
353+
return self.events.insert(events, callback);
354+
}
355+
356+
var tx = {
357+
_id: commitId,
358+
events: events,
359+
aggregateId: events[0].aggregateId,
360+
aggregate: events[0].aggregate,
361+
context: events[0].context
362+
};
363+
364+
self.transactions.insert(tx, function (err) {
339365
if (err) {
340366
debug(err);
341367
if (callback) callback(err);
342368
return;
343369
}
344370

345-
self.removeTransactions(events[events.length - 1]);
371+
self.events.insert(events, function (err) {
372+
if (err) {
373+
debug(err);
374+
if (callback) callback(err);
375+
return;
376+
}
346377

347-
if (callback) { callback(null); }
378+
self.removeTransactions(events[events.length - 1], callback);
379+
});
348380
});
349381
});
350382
},
@@ -641,15 +673,10 @@ _.extend(Mongo.prototype, {
641673

642674
if (evt) {
643675
goodTxs.push(evt);
644-
} else {
645-
self.transactions.remove({ _id: tx._id }, function (err) {
646-
if (err) {
647-
debug(err);
648-
}
649-
});
676+
return clb(null);
650677
}
651-
652-
clb(null);
678+
679+
self.transactions.remove({ _id: tx._id }, clb);
653680
});
654681
}, function (err) {
655682
if (err) {

‎test/storeTest.js‎

Lines changed: 94 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3127,10 +3127,102 @@ types.forEach(function (type) {
31273127

31283128
});
31293129

3130-
});
3130+
if (type === "mongodb" || type === "inmemory") {
3131+
var positionstore = new Store({
3132+
positionsCollectionName: 'positions',
3133+
trackPosition: true,
3134+
});
3135+
3136+
describe('adding position number', function(done) {
3137+
beforeEach(function (done) {
3138+
positionstore.connect(done);
3139+
});
31313140

3141+
afterEach(function (done) {
3142+
positionstore.clear(done);
3143+
});
3144+
describe('with one event in the array', function () {
3145+
3146+
it('it should save the event with position', function(done) {
3147+
3148+
var event = {
3149+
aggregateId: 'id1',
3150+
id: '111',
3151+
streamRevision: 0,
3152+
commitId: '111',
3153+
commitStamp: new Date(),
3154+
commitSequence: 0,
3155+
payload: {
3156+
event:'bla',
3157+
array: []
3158+
},
3159+
applyMappings: function () {}
3160+
};
3161+
3162+
positionstore.addEvents([event], function(err) {
3163+
expect(err).not.to.be.ok();
3164+
3165+
positionstore.getEvents({}, 0, -1, function(err, evts) {
3166+
expect(err).not.to.be.ok();
3167+
expect(evts[0].position).to.eql(1);
3168+
done();
3169+
});
3170+
});
3171+
3172+
});
3173+
3174+
});
3175+
3176+
describe('with multiple events in the array', function () {
3177+
3178+
it('it should save the events with positions', function(done) {
3179+
3180+
var event1 = {
3181+
aggregateId: 'id2',
3182+
streamRevision: 0,
3183+
id: '112',
3184+
commitId: '987',
3185+
commitStamp: new Date(Date.now() + 1),
3186+
commitSequence: 0,
3187+
restInCommitStream: 1,
3188+
payload: {
3189+
event:'bla'
3190+
},
3191+
applyMappings: function () {}
3192+
};
3193+
3194+
var event2 = {
3195+
aggregateId: 'id2',
3196+
streamRevision: 1,
3197+
id:'113',
3198+
commitId: '987',
3199+
commitStamp: new Date(Date.now() + 1),
3200+
commitSequence: 1,
3201+
restInCommitStream: 0,
3202+
payload: {
3203+
event:'bla2'
3204+
},
3205+
applyMappings: function () {}
3206+
};
3207+
3208+
positionstore.addEvents([event1, event2], function(err) {
3209+
expect(err).not.to.be.ok();
3210+
3211+
positionstore.getEvents({}, 0, -1, function(err, evts) {
3212+
expect(err).not.to.be.ok();
3213+
expect(evts).to.be.an('array');
3214+
expect(evts).to.have.length(2);
3215+
expect(evts[0].position).to.eql(1);
3216+
expect(evts[1].position).to.eql(2);
3217+
done();
3218+
});
3219+
});
3220+
});
3221+
});
3222+
});
3223+
}
3224+
});
31323225
});
3133-
31343226
});
31353227

31363228
});

0 commit comments

Comments
 (0)