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); if (!successfulRequests.isEmpty()) { 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 -> { }, now); } else { verify(obj instanceof IncrementSequence); final IncrementSequence increment = (IncrementSequence) obj; successor.doReplayRequest(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(); } 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(); } } 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); @Override final String toString();  }
@Test public void testReplayMessages() { final TestProbe probe = new TestProbe(system); final List<ConnectionEntry> entries = new ArrayList<>(); final Consumer<Response<?, ?>> callback = createCallbackMock(); final ReadTransactionRequest request1 = new ReadTransactionRequest(TRANSACTION_ID, 2L, probe.ref(), PATH_2, true); final ExistsTransactionRequest request2 = new ExistsTransactionRequest(TRANSACTION_ID, 3L, probe.ref(), PATH_3, true); entries.add(AccessClientUtil.createConnectionEntry(request1, callback, 0L)); entries.add(AccessClientUtil.createConnectionEntry(request2, callback, 0L)); final TransactionTester<RemoteProxyTransaction> successor = createRemoteProxyTransactionTester(); final AbortLocalTransactionRequest successful1 = new AbortLocalTransactionRequest(TRANSACTION_ID, probe.ref()); transaction.recordSuccessfulRequest(successful1); final ReadTransactionRequest successful2 = new ReadTransactionRequest(TRANSACTION_ID, 1L, probe.ref(), PATH_1, true); transaction.recordSuccessfulRequest(successful2); transaction.startReconnect(); final ProxyHistory mockSuccessor = mock(ProxyHistory.class); when(mockSuccessor.createTransactionProxy(TRANSACTION_ID, transaction.isSnapshotOnly(), false)) .thenReturn(successor.getTransaction()); transaction.replayMessages(mockSuccessor, entries); final ModifyTransactionRequest transformed = successor.expectTransactionRequest(ModifyTransactionRequest.class); assertNotNull(transformed); assertEquals(successful1.getSequence(), transformed.getSequence()); assertTrue(transformed.getPersistenceProtocol().isPresent()); assertEquals(PersistenceProtocol.ABORT, transformed.getPersistenceProtocol().get()); ReadTransactionRequest tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class); assertNotNull(tmpRead); assertEquals(successful2.getTarget(), tmpRead.getTarget()); assertEquals(successful2.getSequence(), tmpRead.getSequence()); assertEquals(successful2.getPath(), tmpRead.getPath()); assertEquals(successor.localActor(), tmpRead.getReplyTo()); tmpRead = successor.expectTransactionRequest(ReadTransactionRequest.class); assertNotNull(tmpRead); assertEquals(request1.getTarget(), tmpRead.getTarget()); assertEquals(request1.getSequence(), tmpRead.getSequence()); assertEquals(request1.getPath(), tmpRead.getPath()); assertEquals(successor.localActor(), tmpRead.getReplyTo()); final ExistsTransactionRequest tmpExist = successor.expectTransactionRequest(ExistsTransactionRequest.class); assertNotNull(tmpExist); assertEquals(request2.getTarget(), tmpExist.getTarget()); assertEquals(request2.getSequence(), tmpExist.getSequence()); assertEquals(request2.getPath(), tmpExist.getPath()); assertEquals(successor.localActor(), tmpExist.getReplyTo()); }