class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {

    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(), local.isDone());
        LOG.debug("{} created successor {}", 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.doReplayRequest((TransactionRequest<?>) obj, resp -> { /*NOOP*/ }, now);
                } else {
                    verify(obj instanceof IncrementSequence);
                    final IncrementSequence increment = (IncrementSequence) obj;
                    successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
                        increment.getSequence(), localActor(), isSnapshotOnly(),
                        increment.getDelta()), resp -> { /*NOOP*/ }, 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(req instanceof TransactionRequest, "Unhandled request %s", req);
                LOG.debug("Replaying queued request {} to successor {}", req, successor);
                successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
                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.
         */
        if (SEALED.equals(prevState)) {
            LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
            final long enqueuedTicks = parent.currentTime();
            final Optional<ModifyTransactionRequest> optState = flushState();
            if (optState.isPresent()) {
                successor.handleReplayedRemoteRequest(optState.get(), null, enqueuedTicks);
            }
            if (successor.markSealed()) {
                successor.sealAndSend(OptionalLong.of(enqueuedTicks));
            }
        }
    }

      AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone);

}