@Test public void testHandle() { String name = "name"; byte[] badChunk = {8}; FlowFile original = mock(FlowFile.class); FlowFile updated1 = mock(FlowFile.class); FlowFile updated2 = mock(FlowFile.class); FlowFile updated3 = mock(FlowFile.class); FlowFile updated4 = mock(FlowFile.class); ProcessSession session = mock(ProcessSession.class); when(session.create(original)).thenReturn(updated1); when(session.putAttribute(updated1, CoreAttributes.FILENAME.key(), name)).thenReturn(updated2); when(session.putAttribute(updated2, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString())).thenReturn(updated3); ByteArrayOutputStream out = new ByteArrayOutputStream(); when(session.write(eq(updated3), any(OutputStreamCallback.class))).thenAnswer(invocation -> { ((OutputStreamCallback) invocation.getArguments()[1]).process(out); return updated4; }); malformedChunkHandler.handle(original, session, name, badChunk); verify(session).transfer(updated4, badChunkRelationship); assertArrayEquals(badChunk, out.toByteArray()); }
public void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk) { FlowFile flowFile = processSession.create(original); flowFile = processSession.putAttribute(flowFile, CoreAttributes.FILENAME.key(), chunkName); flowFile = processSession.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString()); flowFile = processSession.write(flowFile, out -> out.write(badChunk)); processSession.transfer(flowFile, badChunkRelationship); }
MalformedChunkHandler { public void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk) { FlowFile flowFile = processSession.create(original); flowFile = processSession.putAttribute(flowFile, CoreAttributes.FILENAME.key(), chunkName); flowFile = processSession.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString()); flowFile = processSession.write(flowFile, out -> out.write(badChunk)); processSession.transfer(flowFile, badChunkRelationship); } }
MalformedChunkHandler { public void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk) { FlowFile flowFile = processSession.create(original); flowFile = processSession.putAttribute(flowFile, CoreAttributes.FILENAME.key(), chunkName); flowFile = processSession.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString()); flowFile = processSession.write(flowFile, out -> out.write(badChunk)); processSession.transfer(flowFile, badChunkRelationship); } MalformedChunkHandler(Relationship badChunkRelationship); }
MalformedChunkHandler { public void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk) { FlowFile flowFile = processSession.create(original); flowFile = processSession.putAttribute(flowFile, CoreAttributes.FILENAME.key(), chunkName); flowFile = processSession.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString()); flowFile = processSession.write(flowFile, out -> out.write(badChunk)); processSession.transfer(flowFile, badChunkRelationship); } MalformedChunkHandler(Relationship badChunkRelationship); void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk); }
MalformedChunkHandler { public void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk) { FlowFile flowFile = processSession.create(original); flowFile = processSession.putAttribute(flowFile, CoreAttributes.FILENAME.key(), chunkName); flowFile = processSession.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), MediaType.APPLICATION_BINARY.toString()); flowFile = processSession.write(flowFile, out -> out.write(badChunk)); processSession.transfer(flowFile, badChunkRelationship); } MalformedChunkHandler(Relationship badChunkRelationship); void handle(FlowFile original, ProcessSession processSession, String chunkName, byte[] badChunk);  }