fix(shadow): start queue cursor at database time
This commit is contained in:
@@ -35,4 +35,8 @@ public final class ShadowReadCursor {
|
|||||||
public Instant getLastCreateDate() { return lastCreateDate; }
|
public Instant getLastCreateDate() { return lastCreateDate; }
|
||||||
public Instant getOverlapStart() { return lastCreateDate; }
|
public Instant getOverlapStart() { return lastCreateDate; }
|
||||||
public String getLastGuid() { return lastGuid; }
|
public String getLastGuid() { return lastGuid; }
|
||||||
|
|
||||||
|
public boolean isInitial() {
|
||||||
|
return BEGINNING.equals(lastCreateDate) && lastGuid.isEmpty();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ public final class SqlServerShadowFactReader {
|
|||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
ShadowFactBatch batch = sourceTransaction.execute(status ->
|
ShadowFactBatch batch = sourceTransaction.execute(status ->
|
||||||
collect(cursor.withOverlap(CURSOR_OVERLAP), deadline));
|
collect(cursor, deadline));
|
||||||
if (batch == null) {
|
if (batch == null) {
|
||||||
throw new ShadowFactReadException("SOURCE_EMPTY_BATCH",
|
throw new ShadowFactReadException("SOURCE_EMPTY_BATCH",
|
||||||
Duration.ZERO);
|
Duration.ZERO);
|
||||||
@@ -94,7 +94,11 @@ public final class SqlServerShadowFactReader {
|
|||||||
private ShadowFactBatch collect(ShadowReadCursor cursor, Instant deadline) {
|
private ShadowFactBatch collect(ShadowReadCursor cursor, Instant deadline) {
|
||||||
ensureTime(deadline);
|
ensureTime(deadline);
|
||||||
Instant databaseStartedAt = source.readDatabaseClock(deadline);
|
Instant databaseStartedAt = source.readDatabaseClock(deadline);
|
||||||
List<ShadowQueueFact> queues = immutable(source.readQueues(cursor, deadline));
|
ShadowReadCursor queryCursor = cursor.isInitial()
|
||||||
|
? new ShadowReadCursor(databaseStartedAt, "") : cursor;
|
||||||
|
queryCursor = queryCursor.withOverlap(CURSOR_OVERLAP);
|
||||||
|
List<ShadowQueueFact> queues = immutable(source.readQueues(
|
||||||
|
queryCursor, deadline));
|
||||||
requireLimit("QUEUE", queues.size(), QUEUE_LIMIT);
|
requireLimit("QUEUE", queues.size(), QUEUE_LIMIT);
|
||||||
List<ShadowVehicleFact> vehicles =
|
List<ShadowVehicleFact> vehicles =
|
||||||
immutable(source.readVehicles(deadline));
|
immutable(source.readVehicles(deadline));
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ class SqlServerShadowFactReaderTest {
|
|||||||
siteIds(batch.getMaps()));
|
siteIds(batch.getMaps()));
|
||||||
assertEquals(3, batch.getSourceQueues().size());
|
assertEquals(3, batch.getSourceQueues().size());
|
||||||
assertEquals("070502", batch.vehicle("1").getCurrentSite());
|
assertEquals("070502", batch.vehicle("1").getCurrentSite());
|
||||||
|
assertEquals(NOW.minusMillis(5_500), source.cursor.getOverlapStart());
|
||||||
assertTrue(batch.getDatabaseCapturedAt().isBefore(NOW.plusSeconds(4)));
|
assertTrue(batch.getDatabaseCapturedAt().isBefore(NOW.plusSeconds(4)));
|
||||||
assertEquals(1, transactions.calls);
|
assertEquals(1, transactions.calls);
|
||||||
assertTrue(transactions.readOnly);
|
assertTrue(transactions.readOnly);
|
||||||
|
|||||||
Reference in New Issue
Block a user