MessageSlicer implements AutoCloseable { public void cancelSlicing(final @NonNull Predicate<Identifier> filter) { stateCache.asMap().keySet().removeIf( messageSliceIdentifier -> filter.test(messageSliceIdentifier.getClientIdentifier())); } MessageSlicer(final Builder builder); static Builder builder(); static boolean isHandledMessage(final Object message); boolean slice(final SliceOptions options); boolean handleMessage(final Object message); void checkExpiredSlicedMessageState(); @Override void close(); void cancelSlicing(final @NonNull Predicate<Identifier> filter); static final int DEFAULT_MAX_SLICING_TRIES; }
@Test public void testCancelSlicing() throws IOException { doReturn(1).when(mockInputStream).read(any(byte[].class)); final MessageSlicer slicer = newMessageSlicer("testCloseAllSlicedMessageState", 1); slicer.slice(SliceOptions.builder().identifier(IDENTIFIER).fileBackedOutputStream(mockFiledBackedStream) .sendTo(testProbe.ref()).replyTo(testProbe.ref()).onFailureCallback(mockOnFailureCallback).build()); final FileBackedOutputStream mockFiledBackedStream2 = mock(FileBackedOutputStream.class); setupMockFiledBackedStream(mockFiledBackedStream2); slicer.slice(SliceOptions.builder().identifier(new StringIdentifier("test2")) .fileBackedOutputStream(mockFiledBackedStream2).sendTo(testProbe.ref()).replyTo(testProbe.ref()) .onFailureCallback(mockOnFailureCallback).build()); slicer.cancelSlicing(id -> id.equals(IDENTIFIER)); verify(mockFiledBackedStream).cleanup(); verify(mockFiledBackedStream2, never()).cleanup(); verifyNoMoreInteractions(mockOnFailureCallback); }