@Test public void should_insert_if_not_exist() { offsetRepository.createOrUpdateOffset(1, "default", "Unit", 1); assertThat(mongoCollection.countDocuments()).isEqualTo(1); assertThat(mongoCollection.find()) .extracting("offset", "collection", "strategy", "_tenant") .contains(Tuple.tuple(1L, "Unit", "default", 1)); }
public void createOrUpdateOffset(int tenant, String strategy, String collection, long offset) { Bson offsetFilter = and( eq("_tenant", tenant), eq("strategy", strategy), eq("collection", collection) ); Bson offsetUpdate = Updates.set("offset", offset); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.upsert(true); offerCollection.updateOne(offsetFilter, offsetUpdate, updateOptions); }
OffsetRepository { public void createOrUpdateOffset(int tenant, String strategy, String collection, long offset) { Bson offsetFilter = and( eq("_tenant", tenant), eq("strategy", strategy), eq("collection", collection) ); Bson offsetUpdate = Updates.set("offset", offset); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.upsert(true); offerCollection.updateOne(offsetFilter, offsetUpdate, updateOptions); } }
OffsetRepository { public void createOrUpdateOffset(int tenant, String strategy, String collection, long offset) { Bson offsetFilter = and( eq("_tenant", tenant), eq("strategy", strategy), eq("collection", collection) ); Bson offsetUpdate = Updates.set("offset", offset); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.upsert(true); offerCollection.updateOne(offsetFilter, offsetUpdate, updateOptions); } @VisibleForTesting  OffsetRepository(MongoDbAccess mongoDbAccess, String collectionName);  OffsetRepository(MongoDbAccess mongoDbAccess); }
OffsetRepository { public void createOrUpdateOffset(int tenant, String strategy, String collection, long offset) { Bson offsetFilter = and( eq("_tenant", tenant), eq("strategy", strategy), eq("collection", collection) ); Bson offsetUpdate = Updates.set("offset", offset); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.upsert(true); offerCollection.updateOne(offsetFilter, offsetUpdate, updateOptions); } @VisibleForTesting  OffsetRepository(MongoDbAccess mongoDbAccess, String collectionName);  OffsetRepository(MongoDbAccess mongoDbAccess); void createOrUpdateOffset(int tenant, String strategy, String collection, long offset); long findOffsetBy(int tenant, String strategy, String collection); }
OffsetRepository { public void createOrUpdateOffset(int tenant, String strategy, String collection, long offset) { Bson offsetFilter = and( eq("_tenant", tenant), eq("strategy", strategy), eq("collection", collection) ); Bson offsetUpdate = Updates.set("offset", offset); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.upsert(true); offerCollection.updateOne(offsetFilter, offsetUpdate, updateOptions); } @VisibleForTesting  OffsetRepository(MongoDbAccess mongoDbAccess, String collectionName);  OffsetRepository(MongoDbAccess mongoDbAccess); void createOrUpdateOffset(int tenant, String strategy, String collection, long offset); long findOffsetBy(int tenant, String strategy, String collection); static final String COLLECTION_NAME; }