+ // Called with the connection unlocked
+ final synchronized void startReconnect() {
+ // At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
+ // state. This method is called with the queue still unlocked.
+ final SuccessorState nextState = new SuccessorState();
+ final State prevState = STATE_UPDATER.getAndSet(this, nextState);
+
+ LOG.debug("Start reconnect of proxy {} previous state {}", this, prevState);
+ Verify.verify(!(prevState instanceof SuccessorState), "Proxy %s duplicate reconnect attempt after %s", this,
+ prevState);
+
+ // We have asserted a slow-path state, seal(), canCommit(), directCommit() are forced to slow paths, which will
+ // wait until we unblock nextState's latch before accessing state. Now we record prevState for later use and we
+ // are done.
+ nextState.setPrevState(prevState);
+ }
+
+ // Called with the connection locked
+ final void replayMessages(final AbstractProxyTransaction successor,
+ final Iterable<ConnectionEntry> enqueuedEntries) {
+ final SuccessorState local = getSuccessorState();
+ local.setSuccessor(successor);
+
+ // Replay successful requests first
+ for (Object obj : successfulRequests) {
+ if (obj instanceof TransactionRequest) {
+ LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
+ successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, null);
+ } else {
+ Verify.verify(obj instanceof IncrementSequence);
+ successor.incrementSequence(((IncrementSequence) obj).getDelta());
+ }
+ }
+ LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
+ successfulRequests.clear();
+
+ // Now replay whatever is in the connection
+ final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
+ while (it.hasNext()) {
+ final ConnectionEntry e = it.next();
+ final Request<?, ?> req = e.getRequest();
+
+ if (getIdentifier().equals(req.getTarget())) {
+ Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
+ LOG.debug("Forwarding queued request{} to successor {}", req, successor);
+ successor.handleForwardedRemoteRequest((TransactionRequest<?>) req, e.getCallback());
+ it.remove();
+ }
+ }
+
+ /*
+ * Check the state at which we have started the reconnect attempt. State transitions triggered while we were
+ * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
+ * at the end of this method.
+ */
+ final State prevState = local.getPrevState();
+ if (SEALED.equals(prevState)) {
+ LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
+ flushState(successor);
+ successor.ensureSealed();
+ }
+ }
+
+ // Called with the connection locked
+ final void finishReconnect() {
+ final SuccessorState local = getSuccessorState();
+ LOG.debug("Finishing reconnect of proxy {}", this);
+
+ // All done, release the latch, unblocking seal() and canCommit() slow paths
+ local.finish();
+ }
+
+ /**
+ * Invoked from a retired connection for requests which have been in-flight and need to be re-adjusted
+ * and forwarded to the successor connection.
+ *
+ * @param request Request to be forwarded
+ * @param callback Original callback
+ */
+ final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+ final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
+
+ if (successor instanceof LocalProxyTransaction) {
+ forwardToLocal((LocalProxyTransaction)successor, request, callback);
+ } else if (successor instanceof RemoteProxyTransaction) {
+ forwardToRemote((RemoteProxyTransaction)successor, request, callback);
+ } else {
+ throw new IllegalStateException("Unhandled successor " + successor);
+ }
+ }
+
+ abstract boolean isSnapshotOnly();