MessageSlicer implements AutoCloseable { public boolean slice(final SliceOptions options) { final Identifier identifier = options.getIdentifier(); final Serializable message = options.getMessage(); final FileBackedOutputStream fileBackedStream; if (message != null) { LOG.debug("{}: slice: identifier: {}, message: {}", logContext, identifier, message); requireNonNull(fileBackedStreamFactory, "The FiledBackedStreamFactory must be set in order to call this slice method"); fileBackedStream = fileBackedStreamFactory.newInstance(); try (ObjectOutputStream out = new ObjectOutputStream(fileBackedStream)) { out.writeObject(message); } catch (IOException e) { LOG.debug("{}: Error serializing message for {}", logContext, identifier, e); fileBackedStream.cleanup(); options.getOnFailureCallback().accept(e); return false; } } else { fileBackedStream = options.getFileBackedStream(); } return initializeSlicing(options, fileBackedStream); } 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 testSliceWithFailedSerialization() throws IOException { IOException mockFailure = new IOException("mock IOException"); doThrow(mockFailure).when(mockFiledBackedStream).write(any(byte[].class), anyInt(), anyInt()); doThrow(mockFailure).when(mockFiledBackedStream).write(any(byte[].class)); doThrow(mockFailure).when(mockFiledBackedStream).write(anyInt()); doThrow(mockFailure).when(mockFiledBackedStream).flush(); try (MessageSlicer slicer = newMessageSlicer("testSliceWithFailedSerialization", 100)) { final boolean wasSliced = slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(), mockOnFailureCallback); assertFalse(wasSliced); assertFailureCallback(IOException.class); verify(mockFiledBackedStream).cleanup(); } }
@Test public void testSliceWithByteSourceFailure() throws IOException { IOException mockFailure = new IOException("mock IOException"); doThrow(mockFailure).when(mockByteSource).openStream(); doThrow(mockFailure).when(mockByteSource).openBufferedStream(); try (MessageSlicer slicer = newMessageSlicer("testSliceWithByteSourceFailure", 100)) { final boolean wasSliced = slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(), mockOnFailureCallback); assertFalse(wasSliced); assertFailureCallback(IOException.class); verify(mockFiledBackedStream).cleanup(); } }
@Test public void testSliceWithInputStreamFailure() throws IOException { doReturn(0).when(mockInputStream).read(any(byte[].class)); try (MessageSlicer slicer = newMessageSlicer("testSliceWithInputStreamFailure", 2)) { final boolean wasSliced = slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(), mockOnFailureCallback); assertFalse(wasSliced); assertFailureCallback(IOException.class); verify(mockFiledBackedStream).cleanup(); } }