Support different "since" for source and target
Modify `PersistedModel.replicate` to allow consumers to provide different "since" values to be used for the local and remote changes. Make the "since" filters consistent and include the "since" value in the range via `gte`. Before this commit, the local query used `gt` and the remote query used `gte`.
This commit is contained in:
parent
f857f44cae
commit
2f9cf115c3
|
@ -757,7 +757,7 @@ PersistedModel.changes = function(since, filter, callback) {
|
||||||
|
|
||||||
// TODO(ritch) this whole thing could be optimized a bit more
|
// TODO(ritch) this whole thing could be optimized a bit more
|
||||||
Change.find({ where: {
|
Change.find({ where: {
|
||||||
checkpoint: {gt: since},
|
checkpoint: { gte: since },
|
||||||
modelName: this.modelName
|
modelName: this.modelName
|
||||||
}}, function(err, changes) {
|
}}, function(err, changes) {
|
||||||
if (err) return callback(err);
|
if (err) return callback(err);
|
||||||
|
@ -832,6 +832,10 @@ PersistedModel.replicate = function(since, targetModel, options, callback) {
|
||||||
since = -1;
|
since = -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (typeof since !== 'object') {
|
||||||
|
since = { source: since, target: since };
|
||||||
|
}
|
||||||
|
|
||||||
options = options || {};
|
options = options || {};
|
||||||
|
|
||||||
var sourceModel = this;
|
var sourceModel = this;
|
||||||
|
@ -861,11 +865,11 @@ PersistedModel.replicate = function(since, targetModel, options, callback) {
|
||||||
async.waterfall(tasks, done);
|
async.waterfall(tasks, done);
|
||||||
|
|
||||||
function getSourceChanges(cb) {
|
function getSourceChanges(cb) {
|
||||||
sourceModel.changes(since, options.filter, cb);
|
sourceModel.changes(since.source, options.filter, cb);
|
||||||
}
|
}
|
||||||
|
|
||||||
function getDiffFromTarget(sourceChanges, cb) {
|
function getDiffFromTarget(sourceChanges, cb) {
|
||||||
targetModel.diff(since, sourceChanges, cb);
|
targetModel.diff(since.target, sourceChanges, cb);
|
||||||
}
|
}
|
||||||
|
|
||||||
function createSourceUpdates(_diff, cb) {
|
function createSourceUpdates(_diff, cb) {
|
||||||
|
|
|
@ -8,19 +8,23 @@ var expect = require('chai').expect;
|
||||||
|
|
||||||
describe('Replication / Change APIs', function() {
|
describe('Replication / Change APIs', function() {
|
||||||
var dataSource, SourceModel, TargetModel;
|
var dataSource, SourceModel, TargetModel;
|
||||||
|
var tid = 0; // per-test unique id used e.g. to build unique model names
|
||||||
|
|
||||||
beforeEach(function() {
|
beforeEach(function() {
|
||||||
|
tid++;
|
||||||
var test = this;
|
var test = this;
|
||||||
dataSource = this.dataSource = loopback.createDataSource({
|
dataSource = this.dataSource = loopback.createDataSource({
|
||||||
connector: loopback.Memory
|
connector: loopback.Memory
|
||||||
});
|
});
|
||||||
SourceModel = this.SourceModel = PersistedModel.extend('SourceModel',
|
SourceModel = this.SourceModel = PersistedModel.extend(
|
||||||
|
'SourceModel-' + tid,
|
||||||
{ id: { id: true, type: String, defaultFn: 'guid' } },
|
{ id: { id: true, type: String, defaultFn: 'guid' } },
|
||||||
{ trackChanges: true });
|
{ trackChanges: true });
|
||||||
|
|
||||||
SourceModel.attachTo(dataSource);
|
SourceModel.attachTo(dataSource);
|
||||||
|
|
||||||
TargetModel = this.TargetModel = PersistedModel.extend('TargetModel',
|
TargetModel = this.TargetModel = PersistedModel.extend(
|
||||||
|
'TargetModel-' + tid,
|
||||||
{ id: { id: true, type: String, defaultFn: 'guid' } },
|
{ id: { id: true, type: String, defaultFn: 'guid' } },
|
||||||
{ trackChanges: true });
|
{ trackChanges: true });
|
||||||
|
|
||||||
|
@ -109,6 +113,65 @@ describe('Replication / Change APIs', function() {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('applies "since" filter on source changes', function(done) {
|
||||||
|
async.series([
|
||||||
|
function createModelInSourceCp1(next) {
|
||||||
|
SourceModel.create({ id: '1' }, next);
|
||||||
|
},
|
||||||
|
function checkpoint(next) {
|
||||||
|
SourceModel.checkpoint(next);
|
||||||
|
},
|
||||||
|
function createModelInSourceCp2(next) {
|
||||||
|
SourceModel.create({ id: '2' }, next);
|
||||||
|
},
|
||||||
|
function replicateLastChangeOnly(next) {
|
||||||
|
SourceModel.currentCheckpoint(function(err, cp) {
|
||||||
|
if (err) return done(err);
|
||||||
|
SourceModel.replicate(cp, TargetModel, next);
|
||||||
|
});
|
||||||
|
},
|
||||||
|
function verify(next) {
|
||||||
|
TargetModel.find(function(err, list) {
|
||||||
|
if (err) return done(err);
|
||||||
|
var ids = list.map(function(it) { return it.id; });
|
||||||
|
// '1' should be skipped by replication
|
||||||
|
expect(ids).to.eql(['2']);
|
||||||
|
next();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
], done);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('applies "since" filter on target changes', function(done) {
|
||||||
|
// Because the "since" filter is just an optimization,
|
||||||
|
// there isn't really any observable behaviour we could
|
||||||
|
// check to assert correct implementation.
|
||||||
|
var diffSince = [];
|
||||||
|
spyAndStoreSinceArg(TargetModel, 'diff', diffSince);
|
||||||
|
|
||||||
|
SourceModel.replicate(10, TargetModel, function(err) {
|
||||||
|
if (err) return done(err);
|
||||||
|
expect(diffSince).to.eql([10]);
|
||||||
|
done();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('uses different "since" value for source and target', function(done) {
|
||||||
|
var sourceSince = [];
|
||||||
|
var targetSince = [];
|
||||||
|
|
||||||
|
spyAndStoreSinceArg(SourceModel, 'changes', sourceSince);
|
||||||
|
spyAndStoreSinceArg(TargetModel, 'diff', targetSince);
|
||||||
|
|
||||||
|
var since = { source: 1, target: 2 };
|
||||||
|
SourceModel.replicate(since, TargetModel, function(err) {
|
||||||
|
if (err) return done(err);
|
||||||
|
expect(sourceSince).to.eql([1]);
|
||||||
|
expect(targetSince).to.eql([2]);
|
||||||
|
done();
|
||||||
|
});
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
describe('conflict detection - both updated', function() {
|
describe('conflict detection - both updated', function() {
|
||||||
|
@ -493,4 +556,12 @@ describe('Replication / Change APIs', function() {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
function spyAndStoreSinceArg(Model, methodName, store) {
|
||||||
|
var orig = Model[methodName];
|
||||||
|
Model[methodName] = function(since) {
|
||||||
|
store.push(since);
|
||||||
|
orig.apply(this, arguments);
|
||||||
|
};
|
||||||
|
}
|
||||||
});
|
});
|
||||||
|
|
Loading…
Reference in New Issue