+ * Valid only after first flow has completed. + * + * @return number of records, or -1 if not valid. + */ + public int getInboundCount() { + return numInboundRecords.get(); + } + + /** + * Get the number of records fetched from the second repository (usually the + * local store, hence outbound). + *
+ * Valid only after second flow has completed. + * + * @return number of records, or -1 if not valid. + */ + public int getOutboundCount() { + return numOutboundRecords.get(); + } + + // These are accessed by `abort` and `synchronize`, both of which are synchronized. + // Guarded by `this`. + protected RecordsChannel channelAToB; + protected RecordsChannel channelBToA; + + /** + * Please don't call this until you've been notified with onInitialized. + */ + public synchronized void synchronize() { + numInboundRecords.set(-1); + numOutboundRecords.set(-1); + + // First thing: decide whether we should. + if (sessionA.shouldSkip() || + sessionB.shouldSkip()) { + Logger.info(LOG_TAG, "Session requested skip. Short-circuiting sync."); + sessionA.abort(); + sessionB.abort(); + this.delegate.onSynchronizeSkipped(this); + return; + } + + final SynchronizerSession session = this; + + // TODO: failed record handling. + + // This is the *second* record channel to flow. + // I, SynchronizerSession, am the delegate for the *second* flow. + channelBToA = new RecordsChannel(this.sessionB, this.sessionA, this); + + // This is the delegate for the *first* flow. + RecordsChannelDelegate channelAToBDelegate = new RecordsChannelDelegate() { + public void onFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd) { + session.onFirstFlowCompleted(recordsChannel, fetchEnd, storeEnd); + } + + @Override + public void onFlowBeginFailed(RecordsChannel recordsChannel, Exception ex) { + Logger.warn(LOG_TAG, "First RecordsChannel onFlowBeginFailed. Logging session error.", ex); + session.delegate.onSynchronizeFailed(session, ex, "Failed to begin first flow."); + } + + @Override + public void onFlowFetchFailed(RecordsChannel recordsChannel, Exception ex) { + Logger.warn(LOG_TAG, "First RecordsChannel onFlowFetchFailed. Logging remote fetch error.", ex); + } + + @Override + public void onFlowStoreFailed(RecordsChannel recordsChannel, Exception ex, String recordGuid) { + Logger.warn(LOG_TAG, "First RecordsChannel onFlowStoreFailed. Logging local store error.", ex); + } + + @Override + public void onFlowFinishFailed(RecordsChannel recordsChannel, Exception ex) { + Logger.warn(LOG_TAG, "First RecordsChannel onFlowFinishedFailed. Logging session error.", ex); + session.delegate.onSynchronizeFailed(session, ex, "Failed to finish first flow."); + } + }; + + // This is the *first* channel to flow. + channelAToB = new RecordsChannel(this.sessionA, this.sessionB, channelAToBDelegate); + + Logger.trace(LOG_TAG, "Starting A to B flow. Channel is " + channelAToB); + try { + channelAToB.beginAndFlow(); + } catch (InvalidSessionTransitionException e) { + onFlowBeginFailed(channelAToB, e); + } + } + + /** + * Called after the first flow completes. + *
+ * By default, any fetch and store failures are ignored.
+ * @param recordsChannel the RecordsChannel (for error testing).
+ * @param fetchEnd timestamp when fetches completed.
+ * @param storeEnd timestamp when stores completed.
+ */
+ public void onFirstFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd) {
+ Logger.trace(LOG_TAG, "First RecordsChannel onFlowCompleted.");
+ Logger.debug(LOG_TAG, "Fetch end is " + fetchEnd + ". Store end is " + storeEnd + ". Starting next.");
+ pendingATimestamp = fetchEnd;
+ storeEndBTimestamp = storeEnd;
+ numInboundRecords.set(recordsChannel.getFetchCount());
+ flowAToBCompleted = true;
+ channelBToA.flow();
+ }
+
+ /**
+ * Called after the second flow completes.
+ *
+ * By default, any fetch and store failures are ignored.
+ * @param recordsChannel the RecordsChannel (for error testing).
+ * @param fetchEnd timestamp when fetches completed.
+ * @param storeEnd timestamp when stores completed.
+ */
+ public void onSecondFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd) {
+ Logger.trace(LOG_TAG, "Second RecordsChannel onFlowCompleted.");
+ Logger.debug(LOG_TAG, "Fetch end is " + fetchEnd + ". Store end is " + storeEnd + ". Finishing.");
+
+ pendingBTimestamp = fetchEnd;
+ storeEndATimestamp = storeEnd;
+ numOutboundRecords.set(recordsChannel.getFetchCount());
+ flowBToACompleted = true;
+
+ // Finish the two sessions.
+ try {
+ this.sessionA.finish(this);
+ } catch (InactiveSessionException e) {
+ this.onFinishFailed(e);
+ return;
+ }
+ }
+
+ @Override
+ public void onFlowCompleted(RecordsChannel recordsChannel, long fetchEnd, long storeEnd) {
+ onSecondFlowCompleted(recordsChannel, fetchEnd, storeEnd);
+ }
+
+ @Override
+ public void onFlowBeginFailed(RecordsChannel recordsChannel, Exception ex) {
+ Logger.warn(LOG_TAG, "Second RecordsChannel onFlowBeginFailed. Logging session error.", ex);
+ this.delegate.onSynchronizeFailed(this, ex, "Failed to begin second flow.");
+ }
+
+ @Override
+ public void onFlowFetchFailed(RecordsChannel recordsChannel, Exception ex) {
+ Logger.warn(LOG_TAG, "Second RecordsChannel onFlowFetchFailed. Logging local fetch error.", ex);
+ }
+
+ @Override
+ public void onFlowStoreFailed(RecordsChannel recordsChannel, Exception ex, String recordGuid) {
+ Logger.warn(LOG_TAG, "Second RecordsChannel onFlowStoreFailed. Logging remote store error.", ex);
+ }
+
+ @Override
+ public void onFlowFinishFailed(RecordsChannel recordsChannel, Exception ex) {
+ Logger.warn(LOG_TAG, "Second RecordsChannel onFlowFinishedFailed. Logging session error.", ex);
+ this.delegate.onSynchronizeFailed(this, ex, "Failed to finish second flow.");
+ }
+
+ /*
+ * RepositorySessionCreationDelegate methods.
+ */
+
+ /**
+ * I could be called twice: once for sessionA and once for sessionB.
+ *
+ * I try to clean up sessionA if it is not null, since the creation of
+ * sessionB must have failed.
+ */
+ @Override
+ public void onSessionCreateFailed(Exception ex) {
+ // Attempt to finish the first session, if the second is the one that failed.
+ if (this.sessionA != null) {
+ try {
+ // We no longer need a reference to our context.
+ this.context = null;
+ this.sessionA.finish(this);
+ } catch (Exception e) {
+ // Never mind; best-effort finish.
+ }
+ }
+ // We no longer need a reference to our context.
+ this.context = null;
+ this.delegate.onSynchronizeFailed(this, ex, "Failed to create session");
+ }
+
+ /**
+ * I should be called twice: first for sessionA and second for sessionB.
+ *
+ * If I am called for sessionB, I call my delegate's `onInitialized` callback
+ * method because my repository sessions are correctly initialized.
+ */
+ // TODO: some of this "finish and clean up" code can be refactored out.
+ @Override
+ public void onSessionCreated(RepositorySession session) {
+ if (session == null ||
+ this.sessionA == session) {
+ // TODO: clean up sessionA.
+ this.delegate.onSynchronizeFailed(this, new UnexpectedSessionException(session), "Failed to create session.");
+ return;
+ }
+ if (this.sessionA == null) {
+ this.sessionA = session;
+
+ // Unbundle.
+ try {
+ this.sessionA.unbundle(this.bundleA);
+ } catch (Exception e) {
+ this.delegate.onSynchronizeFailed(this, new UnbundleError(e, sessionA), "Failed to unbundle first session.");
+ // TODO: abort
+ return;
+ }
+ this.getSynchronizer().repositoryB.createSession(this, this.context);
+ return;
+ }
+ if (this.sessionB == null) {
+ this.sessionB = session;
+ // We no longer need a reference to our context.
+ this.context = null;
+
+ // Unbundle. We unbundled sessionA when that session was created.
+ try {
+ this.sessionB.unbundle(this.bundleB);
+ } catch (Exception e) {
+ this.delegate.onSynchronizeFailed(this, new UnbundleError(e, sessionA), "Failed to unbundle second session.");
+ return;
+ }
+
+ this.delegate.onInitialized(this);
+ return;
+ }
+ // TODO: need a way to make sure we don't call any more delegate methods.
+ this.delegate.onSynchronizeFailed(this, new UnexpectedSessionException(session), "Failed to create session.");
+ }
+
+ /*
+ * RepositorySessionFinishDelegate methods.
+ */
+
+ /**
+ * I could be called twice: once for sessionA and once for sessionB.
+ *
+ * If sessionB couldn't be created, I don't fail again.
+ */
+ @Override
+ public void onFinishFailed(Exception ex) {
+ if (this.sessionB == null) {
+ // Ah, it was a problem cleaning up. Never mind.
+ Logger.warn(LOG_TAG, "Got exception cleaning up first after second session creation failed.", ex);
+ return;
+ }
+ String session = (this.sessionA == null) ? "B" : "A";
+ this.delegate.onSynchronizeFailed(this, ex, "Finish of session " + session + " failed.");
+ }
+
+ /**
+ * I should be called twice: first for sessionA and second for sessionB.
+ *
+ * If I am called for sessionA, I try to finish sessionB.
+ *
+ * If I am called for sessionB, I call my delegate's `onSynchronized` callback
+ * method because my flows should have completed.
+ */
+ @Override
+ public void onFinishSucceeded(RepositorySession session,
+ RepositorySessionBundle bundle) {
+ Logger.debug(LOG_TAG, "onFinishSucceeded. Flows? " + flowAToBCompleted + ", " + flowBToACompleted);
+
+ if (session == sessionA) {
+ if (flowAToBCompleted) {
+ Logger.debug(LOG_TAG, "onFinishSucceeded: bumping session A's timestamp to " + pendingATimestamp + " or " + storeEndATimestamp);
+ bundle.bumpTimestamp(Math.max(pendingATimestamp, storeEndATimestamp));
+ this.synchronizer.bundleA = bundle;
+ } else {
+ // Should not happen!
+ this.delegate.onSynchronizeFailed(this, new UnexpectedSessionException(sessionA), "Failed to finish first session.");
+ return;
+ }
+ if (this.sessionB != null) {
+ Logger.trace(LOG_TAG, "Finishing session B.");
+ // On to the next.
+ try {
+ this.sessionB.finish(this);
+ } catch (InactiveSessionException e) {
+ this.onFinishFailed(e);
+ return;
+ }
+ }
+ } else if (session == sessionB) {
+ if (flowBToACompleted) {
+ Logger.debug(LOG_TAG, "onFinishSucceeded: bumping session B's timestamp to " + pendingBTimestamp + " or " + storeEndBTimestamp);
+ bundle.bumpTimestamp(Math.max(pendingBTimestamp, storeEndBTimestamp));
+ this.synchronizer.bundleB = bundle;
+ Logger.trace(LOG_TAG, "Notifying delegate.onSynchronized.");
+ this.delegate.onSynchronized(this);
+ } else {
+ // Should not happen!
+ this.delegate.onSynchronizeFailed(this, new UnexpectedSessionException(sessionB), "Failed to finish second session.");
+ return;
+ }
+ } else {
+ // TODO: hurrrrrr...
+ }
+
+ if (this.sessionB == null) {
+ this.sessionA = null; // We're done.
+ }
+ }
+
+ @Override
+ public RepositorySessionFinishDelegate deferredFinishDelegate(final ExecutorService executor) {
+ return new DeferredRepositorySessionFinishDelegate(this, executor);
+ }
+}