The Tor Browser: comparison mobile/android/base/sync/repositories/RepositorySession.java

--1:000000000000
+:20372ef04d69
+/* This Source Code Form is subject to the terms of the Mozilla Public
+* License, v. 2.0. If a copy of the MPL was not distributed with this
+* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+package org.mozilla.gecko.sync.repositories;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.mozilla.gecko.background.common.log.Logger;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionBeginDelegate;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionFetchRecordsDelegate;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionFinishDelegate;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionGuidsSinceDelegate;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionStoreDelegate;
+import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionWipeDelegate;
+import org.mozilla.gecko.sync.repositories.domain.Record;
+/**
+* A <code>RepositorySession</code> is created and used thusly:
+*
+*<ul>
+* <li>Construct, with a reference to its parent {@link Repository}, by calling
+*   {@link Repository#createSession(RepositorySessionCreationDelegate, android.content.Context)}.</li>
+* <li>Populate with saved information by calling {@link #unbundle(RepositorySessionBundle)}.</li>
+* <li>Begin a sync by calling {@link #begin(RepositorySessionBeginDelegate)}. <code>begin()</code>
+*   is an appropriate place to initialize expensive resources.</li>
+* <li>Perform operations such as {@link #fetchSince(long, RepositorySessionFetchRecordsDelegate)} and
+*   {@link #store(Record)}.</li>
+* <li>Finish by calling {@link #finish(RepositorySessionFinishDelegate)}, retrieving and storing
+*   the current bundle.</li>
+*</ul>
+*
+* If <code>finish()</code> is not called, {@link #abort()} must be called. These calls must
+* <em>always</em> be paired with <code>begin()</code>.
+*
+*/
+public abstract class RepositorySession {
+public enum SessionStatus {
+UNSTARTED,
+ACTIVE,
+ABORTED,
+DONE
+}
+private static final String LOG_TAG = "RepositorySession";
+protected static void trace(String message) {
+Logger.trace(LOG_TAG, message);
+}
+private SessionStatus status = SessionStatus.UNSTARTED;
+protected Repository repository;
+protected RepositorySessionStoreDelegate delegate;
+/**
+* A queue of Runnables which call out into delegates.
+*/
+protected ExecutorService delegateQueue  = Executors.newSingleThreadExecutor();
+/**
+* A queue of Runnables which effect storing.
+* This includes actual store work, and also the consequences of storeDone.
+* This provides strict ordering.
+*/
+protected ExecutorService storeWorkQueue = Executors.newSingleThreadExecutor();
+// The time that the last sync on this collection completed, in milliseconds since epoch.
+private long lastSyncTimestamp = 0;
+public long getLastSyncTimestamp() {
+return lastSyncTimestamp;
+}
+public static long now() {
+return System.currentTimeMillis();
+}
+public RepositorySession(Repository repository) {
+this.repository = repository;
+}
+public abstract void guidsSince(long timestamp, RepositorySessionGuidsSinceDelegate delegate);
+public abstract void fetchSince(long timestamp, RepositorySessionFetchRecordsDelegate delegate);
+public abstract void fetch(String[] guids, RepositorySessionFetchRecordsDelegate delegate) throws InactiveSessionException;
+public abstract void fetchAll(RepositorySessionFetchRecordsDelegate delegate);
+/**
+* Override this if you wish to short-circuit a sync when you know --
+* e.g., by inspecting the database or info/collections -- that no new
+* data are available.
+*
+* @return true if a sync should proceed.
+*/
+public boolean dataAvailable() {
+return true;
+}
+/**
+* @return true if we cannot safely sync from this <code>RepositorySession</code>.
+*/
+public boolean shouldSkip() {
+return false;
+}
+/*
+* Store operations proceed thusly:
+*
+* * Set a delegate
+* * Store an arbitrary number of records. At any time the delegate can be
+*   notified of an error.
+* * Call storeDone to notify the session that no more items are forthcoming.
+* * The store delegate will be notified of error or completion.
+*
+* This arrangement of calls allows for batching at the session level.
+*
+* Store success calls are not guaranteed.
+*/
+public void setStoreDelegate(RepositorySessionStoreDelegate delegate) {
+Logger.debug(LOG_TAG, "Setting store delegate to " + delegate);
+this.delegate = delegate;
+}
+public abstract void store(Record record) throws NoStoreDelegateException;
+public void storeDone() {
+// Our default behavior will be to assume that the Runnable is
+// executed as soon as all the stores synchronously finish, so
+// our end timestamp can just be… now.
+storeDone(now());
+}
+public void storeDone(final long end) {
+Logger.debug(LOG_TAG, "Scheduling onStoreCompleted for after storing is done: " + end);
+Runnable command = new Runnable() {
+@Override
+public void run() {
+delegate.onStoreCompleted(end);
+}
+};
+storeWorkQueue.execute(command);
+}
+public abstract void wipe(RepositorySessionWipeDelegate delegate);
+/**
+* Synchronously perform the shared work of beginning. Throws on failure.
+* @throws InvalidSessionTransitionException
+*
+*/
+protected void sharedBegin() throws InvalidSessionTransitionException {
+Logger.debug(LOG_TAG, "Shared begin.");
+if (delegateQueue.isShutdown()) {
+throw new InvalidSessionTransitionException(null);
+}
+if (storeWorkQueue.isShutdown()) {
+throw new InvalidSessionTransitionException(null);
+}
+this.transitionFrom(SessionStatus.UNSTARTED, SessionStatus.ACTIVE);
+}
+/**
+* Start the session. This is an appropriate place to initialize
+* data access components such as database handles.
+*
+* @param delegate
+* @throws InvalidSessionTransitionException
+*/
+public void begin(RepositorySessionBeginDelegate delegate) throws InvalidSessionTransitionException {
+sharedBegin();
+delegate.deferredBeginDelegate(delegateQueue).onBeginSucceeded(this);
+}
+public void unbundle(RepositorySessionBundle bundle) {
+this.lastSyncTimestamp = bundle == null ? 0 : bundle.getTimestamp();
+}
+/**
+* Override this in your subclasses to return values to save between sessions.
+* Note that RepositorySession automatically bumps the timestamp to the time
+* the last sync began. If unbundled but not begun, this will be the same as the
+* value in the input bundle.
+*
+* The Synchronizer most likely wants to bump the bundle timestamp to be a value
+* return from a fetch call.
+*/
+protected RepositorySessionBundle getBundle() {
+// Why don't we just persist the old bundle?
+long timestamp = getLastSyncTimestamp();
+RepositorySessionBundle bundle = new RepositorySessionBundle(timestamp);
+Logger.debug(LOG_TAG, "Setting bundle timestamp to " + timestamp + ".");
+return bundle;
+}
+/**
+* Just like finish(), but doesn't do any work that should only be performed
+* at the end of a successful sync, and can be called any time.
+*/
+public void abort(RepositorySessionFinishDelegate delegate) {
+this.abort();
+delegate.deferredFinishDelegate(delegateQueue).onFinishSucceeded(this, this.getBundle());
+}
+/**
+* Abnormally terminate the repository session, freeing or closing
+* any resources that were opened during the lifetime of the session.
+*/
+public void abort() {
+// TODO: do something here.
+this.setStatus(SessionStatus.ABORTED);
+try {
+storeWorkQueue.shutdownNow();
+} catch (Exception e) {
+Logger.error(LOG_TAG, "Caught exception shutting down store work queue.", e);
+}
+try {
+delegateQueue.shutdown();
+} catch (Exception e) {
+Logger.error(LOG_TAG, "Caught exception shutting down delegate queue.", e);
+}
+}
+/**
+* End the repository session, freeing or closing any resources
+* that were opened during the lifetime of the session.
+*
+* @param delegate notified of success or failure.
+* @throws InactiveSessionException
+*/
+public void finish(final RepositorySessionFinishDelegate delegate) throws InactiveSessionException {
+try {
+this.transitionFrom(SessionStatus.ACTIVE, SessionStatus.DONE);
+delegate.deferredFinishDelegate(delegateQueue).onFinishSucceeded(this, this.getBundle());
+} catch (InvalidSessionTransitionException e) {
+Logger.error(LOG_TAG, "Tried to finish() an unstarted or already finished session");
+throw new InactiveSessionException(e);
+}
+Logger.trace(LOG_TAG, "Shutting down work queues.");
+storeWorkQueue.shutdown();
+delegateQueue.shutdown();
+}
+/**
+* Run the provided command if we're active and our delegate queue
+* is not shut down.
+*/
+protected synchronized void executeDelegateCommand(Runnable command)
+throws InactiveSessionException {
+if (!isActive() || delegateQueue.isShutdown()) {
+throw new InactiveSessionException(null);
+}
+delegateQueue.execute(command);
+}
+public synchronized void ensureActive() throws InactiveSessionException {
+if (!isActive()) {
+throw new InactiveSessionException(null);
+}
+}
+public synchronized boolean isActive() {
+return status == SessionStatus.ACTIVE;
+}
+public synchronized SessionStatus getStatus() {
+return status;
+}
+public synchronized void setStatus(SessionStatus status) {
+this.status = status;
+}
+public synchronized void transitionFrom(SessionStatus from, SessionStatus to) throws InvalidSessionTransitionException {
+if (from == null || this.status == from) {
+Logger.trace(LOG_TAG, "Successfully transitioning from " + this.status + " to " + to);
+this.status = to;
+return;
+}
+Logger.warn(LOG_TAG, "Wanted to transition from " + from + " but in state " + this.status);
+throw new InvalidSessionTransitionException(null);
+}
+/**
+* Produce a record that is some combination of the remote and local records
+* provided.
+*
+* The returned record must be produced without mutating either remoteRecord
+* or localRecord. It is acceptable to return either remoteRecord or localRecord
+* if no modifications are to be propagated.
+*
+* The returned record *should* have the local androidID and the remote GUID,
+* and some optional merge of data from the two records.
+*
+* This method can be called with records that are identical, or differ in
+* any regard.
+*
+* This method will not be called if:
+*
+* * either record is marked as deleted, or
+* * there is no local mapping for a new remote record.
+*
+* Otherwise, it will be called precisely once.
+*
+* Side-effects (e.g., for transactional storage) can be hooked in here.
+*
+* @param remoteRecord
+*        The record retrieved from upstream, already adjusted for clock skew.
+* @param localRecord
+*        The record retrieved from local storage.
+* @param lastRemoteRetrieval
+*        The timestamp of the last retrieved set of remote records, adjusted for
+*        clock skew.
+* @param lastLocalRetrieval
+*        The timestamp of the last retrieved set of local records.
+* @return
+*        A Record instance to apply, or null to apply nothing.
+*/
+protected Record reconcileRecords(final Record remoteRecord,
+final Record localRecord,
+final long lastRemoteRetrieval,
+final long lastLocalRetrieval) {
+Logger.debug(LOG_TAG, "Reconciling remote " + remoteRecord.guid + " against local " + localRecord.guid);
+if (localRecord.equalPayloads(remoteRecord)) {
+if (remoteRecord.lastModified > localRecord.lastModified) {
+Logger.debug(LOG_TAG, "Records are equal. No record application needed.");
+return null;
+}
+// Local wins.
+return null;
+}
+// TODO: Decide what to do based on:
+// * Which of the two records is modified;
+// * Whether they are equal or congruent;
+// * The modified times of each record (interpreted through the lens of clock skew);
+// * ...
+boolean localIsMoreRecent = localRecord.lastModified > remoteRecord.lastModified;
+Logger.debug(LOG_TAG, "Local record is more recent? " + localIsMoreRecent);
+Record donor = localIsMoreRecent ? localRecord : remoteRecord;
+// Modify the local record to match the remote record's GUID and values.
+// Preserve the local Android ID, and merge data where possible.
+// It sure would be nice if copyWithIDs didn't give a shit about androidID, mm?
+Record out = donor.copyWithIDs(remoteRecord.guid, localRecord.androidID);
+// We don't want to upload the record if the remote record was
+// applied without changes.
+// This logic will become more complicated as reconciling becomes smarter.
+if (!localIsMoreRecent) {
+trackGUID(out.guid);
+}
+return out;
+}
+/**
+* Depending on the RepositorySession implementation, track
+* that a record — most likely a brand-new record that has been
+* applied unmodified — should be tracked so as to not be uploaded
+* redundantly.
+*
+* The default implementations do nothing.
+*/
+protected void trackGUID(String guid) {
+}
+protected synchronized void untrackGUIDs(Collection<String> guids) {
+}
+protected void untrackGUID(String guid) {
+}
+// Ah, Java. You wretched creature.
+public Iterator<String> getTrackedRecordIDs() {
+return new ArrayList<String>().iterator();
+}
+}

The Tor Browser / file comparison

comparison: mobile/android/base/sync/repositories/RepositorySession.java

mobile/android/base/sync/repositories/RepositorySession.java