@Test void requestMoreBlocks_shouldInvokeCallbackWhenRequestFails() { final Runnable callback = mock(Runnable.class); final Batch batch = createBatch(5, 2); batch.requestMoreBlocks(callback); requestError(batch, new RuntimeException("Oops")); verify(callback).run(); }
@Override public void requestMoreBlocks(final Runnable callback) { checkState( !isComplete() || isContested(), "Attempting to request more blocks from a complete batch"); final RequestHandler requestHandler = new RequestHandler(); final UInt64 startSlot = getLastBlock().map(SignedBeaconBlock::getSlot).map(UInt64::increment).orElse(firstSlot); final UInt64 remainingSlots = count.minus(startSlot.minus(firstSlot)); checkState( remainingSlots.isGreaterThan(UInt64.ZERO), "Attempting to request more blocks when block for last slot already present."); if (currentSyncSource.isEmpty()) { currentSyncSource = syncSourceProvider.selectSource(); if (currentSyncSource.isEmpty()) { eventThread.executeLater(callback); return; } } awaitingBlocks = true; final SyncSource syncSource = currentSyncSource.orElseThrow(); LOG.debug( "Requesting {} slots starting at {} from peer {}", remainingSlots, startSlot, syncSource); syncSource .requestBlocksByRange(startSlot, remainingSlots, UInt64.ONE, requestHandler) .thenRunAsync(() -> onRequestComplete(requestHandler), eventThread) .handleAsync( (__, error) -> { if (error != null) { handleRequestErrors(error); } eventThread.executeLater(callback); return null; }, eventThread) .reportExceptions(); }
SyncSourceBatch implements Batch { @Override public void requestMoreBlocks(final Runnable callback) { checkState( !isComplete() || isContested(), "Attempting to request more blocks from a complete batch"); final RequestHandler requestHandler = new RequestHandler(); final UInt64 startSlot = getLastBlock().map(SignedBeaconBlock::getSlot).map(UInt64::increment).orElse(firstSlot); final UInt64 remainingSlots = count.minus(startSlot.minus(firstSlot)); checkState( remainingSlots.isGreaterThan(UInt64.ZERO), "Attempting to request more blocks when block for last slot already present."); if (currentSyncSource.isEmpty()) { currentSyncSource = syncSourceProvider.selectSource(); if (currentSyncSource.isEmpty()) { eventThread.executeLater(callback); return; } } awaitingBlocks = true; final SyncSource syncSource = currentSyncSource.orElseThrow(); LOG.debug( "Requesting {} slots starting at {} from peer {}", remainingSlots, startSlot, syncSource); syncSource .requestBlocksByRange(startSlot, remainingSlots, UInt64.ONE, requestHandler) .thenRunAsync(() -> onRequestComplete(requestHandler), eventThread) .handleAsync( (__, error) -> { if (error != null) { handleRequestErrors(error); } eventThread.executeLater(callback); return null; }, eventThread) .reportExceptions(); } }
SyncSourceBatch implements Batch { @Override public void requestMoreBlocks(final Runnable callback) { checkState( !isComplete() || isContested(), "Attempting to request more blocks from a complete batch"); final RequestHandler requestHandler = new RequestHandler(); final UInt64 startSlot = getLastBlock().map(SignedBeaconBlock::getSlot).map(UInt64::increment).orElse(firstSlot); final UInt64 remainingSlots = count.minus(startSlot.minus(firstSlot)); checkState( remainingSlots.isGreaterThan(UInt64.ZERO), "Attempting to request more blocks when block for last slot already present."); if (currentSyncSource.isEmpty()) { currentSyncSource = syncSourceProvider.selectSource(); if (currentSyncSource.isEmpty()) { eventThread.executeLater(callback); return; } } awaitingBlocks = true; final SyncSource syncSource = currentSyncSource.orElseThrow(); LOG.debug( "Requesting {} slots starting at {} from peer {}", remainingSlots, startSlot, syncSource); syncSource .requestBlocksByRange(startSlot, remainingSlots, UInt64.ONE, requestHandler) .thenRunAsync(() -> onRequestComplete(requestHandler), eventThread) .handleAsync( (__, error) -> { if (error != null) { handleRequestErrors(error); } eventThread.executeLater(callback); return null; }, eventThread) .reportExceptions(); } SyncSourceBatch(
      final EventThread eventThread,
      final SyncSourceSelector syncSourceProvider,
      final ConflictResolutionStrategy conflictResolutionStrategy,
      final TargetChain targetChain,
      final UInt64 firstSlot,
      final UInt64 count); }
SyncSourceBatch implements Batch { @Override public void requestMoreBlocks(final Runnable callback) { checkState( !isComplete() || isContested(), "Attempting to request more blocks from a complete batch"); final RequestHandler requestHandler = new RequestHandler(); final UInt64 startSlot = getLastBlock().map(SignedBeaconBlock::getSlot).map(UInt64::increment).orElse(firstSlot); final UInt64 remainingSlots = count.minus(startSlot.minus(firstSlot)); checkState( remainingSlots.isGreaterThan(UInt64.ZERO), "Attempting to request more blocks when block for last slot already present."); if (currentSyncSource.isEmpty()) { currentSyncSource = syncSourceProvider.selectSource(); if (currentSyncSource.isEmpty()) { eventThread.executeLater(callback); return; } } awaitingBlocks = true; final SyncSource syncSource = currentSyncSource.orElseThrow(); LOG.debug( "Requesting {} slots starting at {} from peer {}", remainingSlots, startSlot, syncSource); syncSource .requestBlocksByRange(startSlot, remainingSlots, UInt64.ONE, requestHandler) .thenRunAsync(() -> onRequestComplete(requestHandler), eventThread) .handleAsync( (__, error) -> { if (error != null) { handleRequestErrors(error); } eventThread.executeLater(callback); return null; }, eventThread) .reportExceptions(); } SyncSourceBatch(
      final EventThread eventThread,
      final SyncSourceSelector syncSourceProvider,
      final ConflictResolutionStrategy conflictResolutionStrategy,
      final TargetChain targetChain,
      final UInt64 firstSlot,
      final UInt64 count); @Override UInt64 getFirstSlot(); @Override UInt64 getLastSlot(); @Override Optional<SignedBeaconBlock> getFirstBlock(); @Override Optional<SignedBeaconBlock> getLastBlock(); @Override List<SignedBeaconBlock> getBlocks(); @Override void markComplete(); @Override boolean isComplete(); @Override boolean isConfirmed(); @Override boolean isFirstBlockConfirmed(); @Override boolean isContested(); @Override boolean isEmpty(); @Override boolean isAwaitingBlocks(); @Override void markFirstBlockConfirmed(); @Override void markLastBlockConfirmed(); @Override void markAsContested(); @Override void markAsInvalid(); @Override void requestMoreBlocks(final Runnable callback); @Override TargetChain getTargetChain(); @Override String toString(); }
SyncSourceBatch implements Batch { @Override public void requestMoreBlocks(final Runnable callback) { checkState( !isComplete() || isContested(), "Attempting to request more blocks from a complete batch"); final RequestHandler requestHandler = new RequestHandler(); final UInt64 startSlot = getLastBlock().map(SignedBeaconBlock::getSlot).map(UInt64::increment).orElse(firstSlot); final UInt64 remainingSlots = count.minus(startSlot.minus(firstSlot)); checkState( remainingSlots.isGreaterThan(UInt64.ZERO), "Attempting to request more blocks when block for last slot already present."); if (currentSyncSource.isEmpty()) { currentSyncSource = syncSourceProvider.selectSource(); if (currentSyncSource.isEmpty()) { eventThread.executeLater(callback); return; } } awaitingBlocks = true; final SyncSource syncSource = currentSyncSource.orElseThrow(); LOG.debug( "Requesting {} slots starting at {} from peer {}", remainingSlots, startSlot, syncSource); syncSource .requestBlocksByRange(startSlot, remainingSlots, UInt64.ONE, requestHandler) .thenRunAsync(() -> onRequestComplete(requestHandler), eventThread) .handleAsync( (__, error) -> { if (error != null) { handleRequestErrors(error); } eventThread.executeLater(callback); return null; }, eventThread) .reportExceptions(); } SyncSourceBatch(
      final EventThread eventThread,
      final SyncSourceSelector syncSourceProvider,
      final ConflictResolutionStrategy conflictResolutionStrategy,
      final TargetChain targetChain,
      final UInt64 firstSlot,
      final UInt64 count); @Override UInt64 getFirstSlot(); @Override UInt64 getLastSlot(); @Override Optional<SignedBeaconBlock> getFirstBlock(); @Override Optional<SignedBeaconBlock> getLastBlock(); @Override List<SignedBeaconBlock> getBlocks(); @Override void markComplete(); @Override boolean isComplete(); @Override boolean isConfirmed(); @Override boolean isFirstBlockConfirmed(); @Override boolean isContested(); @Override boolean isEmpty(); @Override boolean isAwaitingBlocks(); @Override void markFirstBlockConfirmed(); @Override void markLastBlockConfirmed(); @Override void markAsContested(); @Override void markAsInvalid(); @Override void requestMoreBlocks(final Runnable callback); @Override TargetChain getTargetChain(); @Override String toString();  }