diff --git a/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataCollection.java b/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataCollection.java index e02c93e0ec0..9e86b0de57d 100755 --- a/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataCollection.java +++ b/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataCollection.java @@ -2,7 +2,6 @@ package com.sap.sailing.domain.persistence.racelog.tracking.impl; import java.util.HashMap; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Logger; import org.bson.Document; @@ -43,7 +42,7 @@ import com.sap.sse.common.TypeBasedServiceFinder; public class MetadataCollection extends MongoFixHandler { private static final Logger logger = Logger.getLogger(MetadataCollection.class.getName()); private final MongoCollection metadataCollection; - private final ConcurrentHashMap metadataUpdaters; + private final HashMap metadataUpdaters; private final WriteConcern writeConcern; private final ReadConcern readConcern; @@ -52,7 +51,7 @@ public class MetadataCollection extends MongoFixHandler { TypeBasedServiceFinder deviceServiceFinder, ReadConcern readConcern, WriteConcern writeConcern) { super(fixServiceFinder, deviceServiceFinder); this.metadataCollection = mongoOF.getGPSFixMetadataCollection(); - this.metadataUpdaters = new ConcurrentHashMap<>(); + this.metadataUpdaters = new HashMap<>(); this.readConcern = readConcern; this.writeConcern = writeConcern; } @@ -131,13 +130,13 @@ public class MetadataCollection extends MongoFixHandler { return result; } - void enqueueMetadataUpdate(DeviceIdentifier device, final Object dbDeviceId, + synchronized void enqueueMetadataUpdate(DeviceIdentifier device, final Object dbDeviceId, final int nrOfTotalFixes, TimeRange fixesTimeRange, FixT latestFix) throws TransformationException { final MetadataUpdater metadataUpdaterForDevice = metadataUpdaters.computeIfAbsent(device, d->new MetadataUpdater(this, device)); metadataUpdaterForDevice.enqueueMetadataUpdate(device, dbDeviceId, nrOfTotalFixes, fixesTimeRange, latestFix); } - private void waitForPendingMetadataUpdates(DeviceIdentifier device) { + private synchronized void waitForPendingMetadataUpdates(DeviceIdentifier device) { final MetadataUpdater metadataUpdaterForDevice = metadataUpdaters.get(device); if (metadataUpdaterForDevice != null) { metadataUpdaterForDevice.waitForPendingUpdates(); @@ -157,6 +156,7 @@ public class MetadataCollection extends MongoFixHandler { MongoObjectFactoryImpl.storeTimeRange(newTimeRange, newMetadata, FieldNames.TIMERANGE); updateOperation.append("$set", newMetadata); updateOperation.append("$inc", new Document(FieldNames.NUM_FIXES.name(), update.getNrOfTotalFixes())); + logger.fine(()->"Updating sensor fix store metadata with update operation "+updateOperation); metadataCollection.withWriteConcern(writeConcern).updateOne(com.sap.sailing.shared.persistence.impl.MongoObjectFactoryImpl.getDeviceQuery(deviceServiceFinder, update.getDevice()), updateOperation, new UpdateOptions().upsert(true)); } } diff --git a/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataUpdater.java b/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataUpdater.java index 67d705b2c79..508fbe9fe46 100755 --- a/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataUpdater.java +++ b/java/com.sap.sailing.domain.persistence/src/com/sap/sailing/domain/persistence/racelog/tracking/impl/MetadataUpdater.java @@ -96,8 +96,6 @@ public class MetadataUpdater { } else { setNextUpdate(null); } - updatesProcessed++; - MetadataUpdater.this.notifyAll(); } if (theNextUpdate != null) { boolean success = false; @@ -106,6 +104,10 @@ public class MetadataUpdater { try { metadataCollection.update(theNextUpdate); success = true; + synchronized (MetadataUpdater.this) { + updatesProcessed++; + MetadataUpdater.this.notifyAll(); + } } catch (Exception e) { logger.severe("Unable to write update "+theNextUpdate+" to the metadata collection for device "+forDevice+ ": "+e.getMessage()+"; retrying "+retryCount+" more times."); diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter.test/META-INF/MANIFEST.MF b/java/com.sap.sailing.domain.racelogtrackingadapter.test/META-INF/MANIFEST.MF index 4cb97af7398..d9f03889ea8 100644 --- a/java/com.sap.sailing.domain.racelogtrackingadapter.test/META-INF/MANIFEST.MF +++ b/java/com.sap.sailing.domain.racelogtrackingadapter.test/META-INF/MANIFEST.MF @@ -12,7 +12,10 @@ Require-Bundle: org.junit;bundle-version="4.8.2", org.objenesis;bundle-version="1.3.0", org.hamcrest;bundle-version="1.1.0", com.sap.sse.common, - com.sap.sailing.domain + com.sap.sailing.domain, + org.mongodb.driver-core, + com.sap.sse.mongodb, + org.mongodb.driver-sync Import-Package: com.sap.sailing.domain.abstractlog.regatta.events, com.sap.sailing.domain.persistence.impl, com.sap.sailing.domain.persistence.racelog.tracking.impl, diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/AbstractGPSFixStoreTest.java b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/AbstractGPSFixStoreTest.java index 6e076cc0d08..83b5333a959 100644 --- a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/AbstractGPSFixStoreTest.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/AbstractGPSFixStoreTest.java @@ -12,6 +12,7 @@ import org.junit.Before; import com.mongodb.ReadConcern; import com.mongodb.WriteConcern; +import com.mongodb.client.ClientSession; import com.sap.sailing.domain.abstractlog.race.impl.RaceLogImpl; import com.sap.sailing.domain.abstractlog.regatta.RegattaLog; import com.sap.sailing.domain.abstractlog.regatta.events.impl.RegattaLogDefineMarkEventImpl; @@ -54,6 +55,7 @@ import com.sap.sailing.server.impl.RacingEventServiceImpl; import com.sap.sailing.server.interfaces.RacingEventService; import com.sap.sse.common.impl.DegreeBearingImpl; import com.sap.sse.common.impl.MillisecondsTimePoint; +import com.sap.sse.mongodb.MongoDBService; public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper { protected RacingEventService service; @@ -67,6 +69,7 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper { protected final Competitor comp = DomainFactory.INSTANCE.getOrCreateCompetitor("comp", "comp", null, null, null, null, null, /* timeOnTimeFactor */ null, /* timeOnDistanceAllowanceInSecondsPerNauticalMile */ null, null, /* storePersistently */ true); protected final Boat boat = DomainFactory.INSTANCE.getOrCreateBoat("boat", "boat", boatClass, "GER 234", null, /* storePersistently */ true); protected final Mark mark = DomainFactory.INSTANCE.getOrCreateMark("mark"); + private ClientSession clientSession; protected GPSFixMoving createFix(long millis, double lat, double lng, double knots, double degrees) { return new GPSFixMovingImpl(new DegreePosition(lat, lng), @@ -84,6 +87,7 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper { raceLog = new RaceLogImpl("racelog"); regattaLog = new RegattaLogImpl("regattalog"); dropPersistedData(); + clientSession = MongoDBService.INSTANCE.startCausallyConsistentSession(); store = new MongoSensorFixStoreImpl(service.getMongoObjectFactory(), service.getDomainObjectFactory(), serviceFinderFactory, ReadConcern.MAJORITY, WriteConcern.MAJORITY); } @@ -91,14 +95,15 @@ public class AbstractGPSFixStoreTest extends RaceLogTrackingTestHelper { @After public void after() { dropPersistedData(); + clientSession.close(); } private void dropPersistedData() { MongoObjectFactoryImpl mongoOF = (MongoObjectFactoryImpl) service.getMongoObjectFactory(); - mongoOF.getGPSFixCollection().drop(); - mongoOF.getGPSFixMetadataCollection().drop(); - mongoOF.getRaceLogCollection().drop(); - mongoOF.getRegattaLogCollection().drop(); + mongoOF.getGPSFixCollection().withWriteConcern(WriteConcern.MAJORITY).drop(); + mongoOF.getGPSFixMetadataCollection().withWriteConcern(WriteConcern.MAJORITY).drop(); + mongoOF.getRaceLogCollection().withWriteConcern(WriteConcern.MAJORITY).drop(); + mongoOF.getRegattaLogCollection().withWriteConcern(WriteConcern.MAJORITY).drop(); } protected void map(RegattaLog regattaLog, Competitor comp, DeviceIdentifier device, long from, long to) { diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/SensorFixStoreAndLoadTest.java b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/SensorFixStoreAndLoadTest.java index 2541d56e3a9..cdcde9bcbe5 100644 --- a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/SensorFixStoreAndLoadTest.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/SensorFixStoreAndLoadTest.java @@ -447,7 +447,7 @@ public class SensorFixStoreAndLoadTest { } throw new IllegalArgumentException("Unknown event type"); } - }); + }, /* removeOutliersFromCompetitorTracks */ false); } protected void testNumberOfRawFixes(Track track, long expected) { diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/TrackedRaceLoadsFixesTest.java b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/TrackedRaceLoadsFixesTest.java index 9dba4b06521..f4a7a6bd275 100755 --- a/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/TrackedRaceLoadsFixesTest.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter.test/src/com/sap/sailing/domain/racelogtracking/test/impl/TrackedRaceLoadsFixesTest.java @@ -90,7 +90,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest { final DynamicTrackedRaceImpl trackedRace = createDynamicTrackedRace(boatClass, raceDefinition); trackedRace.setStartOfTrackingReceived(new MillisecondsTimePoint(1000)); trackedRace.setEndOfTrackingReceived(new MillisecondsTimePoint(2000)); - new FixLoaderAndTracker(trackedRace, store, null); + new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false); trackedRace.attachRaceLog(raceLog); trackedRace.attachRegattaLog(regattaLog); @@ -134,7 +134,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest { DynamicTrackedRace trackedRace = createDynamicTrackedRace(boatClass, raceDefinition); - new FixLoaderAndTracker(trackedRace, store, null); + new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false); raceLog.add(new RaceLogStartOfTrackingEventImpl(TimePoint.BeginningOfTime, author, 0)); trackedRace.attachRaceLog(raceLog); @@ -529,7 +529,7 @@ public class TrackedRaceLoadsFixesTest extends AbstractGPSFixStoreTest { DynamicTrackedRace trackedRace = createDynamicTrackedRace(boatClass, raceDefinition); trackedRace.attachRaceLog(raceLog); trackedRace.attachRegattaLog(regattaLog); - new FixLoaderAndTracker(trackedRace, store, null); + new FixLoaderAndTracker(trackedRace, store, null, /* removeOutliersFromCompetitorTracks */ false); for(Consumer afterTrackingStarted : afterTrackingStartedConsumers) { trackedRace.waitForLoadingToFinish(); afterTrackingStarted.accept(trackedRace); diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/FixLoaderAndTracker.java b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/FixLoaderAndTracker.java index 0c389b7d8bb..eff1d0b76f4 100755 --- a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/FixLoaderAndTracker.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/FixLoaderAndTracker.java @@ -154,6 +154,12 @@ public class FixLoaderAndTracker implements TrackingDataLoader { private final SensorFixMapperFactory sensorFixMapperFactory; + /** + * If set to {@code true} in the constructor, fixes loaded for competitor tracks will be subject to + * outlier removal using the {@link OutlierFilter}. + */ + private final boolean removeOutliersFromCompetitorTracks; + /** * This flag is used to tell the loaders/trackers whether preemptive stopping has been requested. If switched * to {@code true}, running loaders will stop loading fixes and return immediately. @@ -168,6 +174,7 @@ public class FixLoaderAndTracker implements TrackingDataLoader { private AtomicBoolean willBeRemovedAfterStopping = new AtomicBoolean(false); private AtomicBoolean stopRequested = new AtomicBoolean(false); + private final AbstractRaceChangeListener raceChangeListener = new AbstractRaceChangeListener() { @Override public void startOfTrackingChanged(TimePoint oldStartOfTracking, TimePoint newStartOfTracking) { @@ -196,6 +203,7 @@ public class FixLoaderAndTracker implements TrackingDataLoader { deviceMappings.addRegattaLog(regattaLog); } }; + private final FixReceivedListener listener = new FixReceivedListener() { @Override public Iterable> fixReceived(DeviceIdentifier device, Timed fix, boolean returnManeuverChanges, boolean returnLiveDelay) { @@ -394,10 +402,11 @@ public class FixLoaderAndTracker implements TrackingDataLoader { } public FixLoaderAndTracker(DynamicTrackedRace trackedRace, SensorFixStore sensorFixStore, - SensorFixMapperFactory sensorFixMapperFactory) { + SensorFixMapperFactory sensorFixMapperFactory, boolean removeOutliersFromCompetitorTracks) { this.sensorFixStore = sensorFixStore; this.sensorFixMapperFactory = sensorFixMapperFactory; this.trackedRace = trackedRace; + this.removeOutliersFromCompetitorTracks = removeOutliersFromCompetitorTracks; startTracking(); } @@ -561,8 +570,6 @@ public class FixLoaderAndTracker implements TrackingDataLoader { /** * First loads the fixes into a temporary track which is then subject to outlier filtering (see * {@link OutlierFilter}). The fixes that make it through outlier filtering are then inserted - * @param competitor - * @param event */ private void loadForCompetitor(Competitor competitor, RegattaLogDeviceMappingEvent event) { DynamicGPSFixTrack track = trackedRace.getTrack(competitor); @@ -574,17 +581,22 @@ public class FixLoaderAndTracker implements TrackingDataLoader { new DynamicGPSFixMovingTrackImpl(track.getTrackedItem(), track.getMillisecondsOverWhichToAverageSpeed()); loadedFixes.suspendValidityAndMaxSpeedCaching(); // no validity nor max speed required on this track try { + final DynamicGPSFixTrack filteredTrack; sensorFixStore. loadFixes(fix -> loadedFixes.add(fix, true), event.getDevice(), timeRangeToLoad.from(), timeRangeToLoad.to(), /* toIsInclusive */ false, stopCallback, progressConsumer); - final Pair> filtered = new OutlierFilter().findAndRemoveInconsistenciesOnRawFixes(loadedFixes); - loadedFixes.lockForRead(); - try { - logger.info("Filtered competitor track for outliers; "+filtered.getA()+" outliers removed in track with "+Util.size(loadedFixes.getRawFixes())+" fixes"); - } finally { - loadedFixes.unlockAfterRead(); + if (removeOutliersFromCompetitorTracks) { + final Pair> filtered = new OutlierFilter().findAndRemoveInconsistenciesOnRawFixes(loadedFixes); + loadedFixes.lockForRead(); + try { + logger.info("Filtered competitor track for outliers; "+filtered.getA()+" outliers removed in track with "+Util.size(loadedFixes.getRawFixes())+" fixes"); + } finally { + loadedFixes.unlockAfterRead(); + } + filteredTrack = filtered.getB(); + } else { + filteredTrack = loadedFixes; } - final DynamicGPSFixTrack filteredTrack = filtered.getB(); track.suspendValidityAndMaxSpeedCaching(); filteredTrack.lockForRead(); try { diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RaceLogFixTrackerManager.java b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RaceLogFixTrackerManager.java index 4ba1228096d..c7770ac2047 100644 --- a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RaceLogFixTrackerManager.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RaceLogFixTrackerManager.java @@ -128,7 +128,7 @@ public class RaceLogFixTrackerManager implements TrackingDataLoader { private synchronized void startTrackerIfNotAlreadyStarted() { if (tracker == null) { logger.fine("Starting fix tracker for TrackedRace: " + trackedRace.getRaceIdentifier()); - tracker = new FixLoaderAndTracker(trackedRace, sensorFixStore, sensorFixMapperFactory); + tracker = new FixLoaderAndTracker(trackedRace, sensorFixStore, sensorFixMapperFactory, /* removeOutliersFromCompetitorTracks */ true); } } diff --git a/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/MongoDBService.java b/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/MongoDBService.java index 55160b1c355..663c2aa6509 100644 --- a/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/MongoDBService.java +++ b/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/MongoDBService.java @@ -49,4 +49,6 @@ public interface MongoDBService { * Those weak references that got cleared will be removed from the cache. */ ClientSession startAutoRefreshingSession(); + + ClientSession startCausallyConsistentSession(); } diff --git a/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/internal/MongoDBServiceImpl.java b/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/internal/MongoDBServiceImpl.java index 3726bb08153..644ffbde24c 100644 --- a/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/internal/MongoDBServiceImpl.java +++ b/java/com.sap.sse.mongodb/src/com/sap/sse/mongodb/internal/MongoDBServiceImpl.java @@ -13,6 +13,7 @@ import java.util.logging.Logger; import org.bson.BsonArray; import org.bson.BsonDocument; +import com.mongodb.ClientSessionOptions; import com.mongodb.ConnectionString; import com.mongodb.client.ClientSession; import com.mongodb.client.MongoClient; @@ -72,6 +73,12 @@ public class MongoDBServiceImpl implements MongoDBService { } } + @Override + public ClientSession startCausallyConsistentSession() { + ensureConfigurationDefaultingToTest(); + return getMongo(getConfiguration()).startSession(ClientSessionOptions.builder().causallyConsistent(true).build()); + } + @Override public ClientSession startAutoRefreshingSession() { ensureConfigurationDefaultingToTest(); diff --git a/java/target/configuration/logging_debug.properties b/java/target/configuration/logging_debug.properties index a56b7e6c981..a75b14a2c80 100644 --- a/java/target/configuration/logging_debug.properties +++ b/java/target/configuration/logging_debug.properties @@ -49,3 +49,5 @@ com.sap.sse.replication.impl.OperationQueueByKeyExecutor.level = INFO # YB logging com.sap.sailing.domain.yellowbrickadapter.level = ALL + +com.sap.sailing.domain.persistence.racelog.tracking.impl.MetadataCollection.level = FINE \ No newline at end of file