+ // Called with the connection locked
+ final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
+ final SuccessorState local = getSuccessorState();
+ final State prevState = local.getPrevState();
+
+ final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
+ isSnapshotOnly());
+ LOG.debug("{} created successor transaction proxy {}", this, successor);
+ local.setSuccessor(successor);
+
+ // Replay successful requests first
+ if (!successfulRequests.isEmpty()) {
+ // We need to find a good timestamp to use for successful requests, as we do not want to time them out
+ // nor create timing inconsistencies in the queue -- requests are expected to be ordered by their enqueue
+ // time. We will pick the time of the first entry available. If there is none, we will just use current
+ // time, as all other requests will get enqueued afterwards.
+ final ConnectionEntry firstInQueue = Iterables.getFirst(enqueuedEntries, null);
+ final long now = firstInQueue != null ? firstInQueue.getEnqueuedTicks() : parent.currentTime();
+
+ for (Object obj : successfulRequests) {
+ if (obj instanceof TransactionRequest) {
+ LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
+ successor.replayRequest((TransactionRequest<?>) obj, resp -> { }, now);
+ } else {
+ Verify.verify(obj instanceof IncrementSequence);
+ final IncrementSequence increment = (IncrementSequence) obj;
+ successor.replayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
+ increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
+ now);
+ LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
+ }
+ }
+ 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("Replaying queued request {} to successor {}", req, successor);
+ successor.replayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
+ it.remove();
+ }
+ }