@Test public void testAppend() throws IOException { FlowFile ff = session.create(); ff = session.append(ff, new OutputStreamCallback() { @Override public void process(OutputStream out) throws IOException { out.write("Hello".getBytes()); } }); contentRepo.disableRead = true; ff = session.append(ff, new OutputStreamCallback() { @Override public void process(OutputStream out) throws IOException { out.write(", ".getBytes()); } }); ff = session.append(ff, new OutputStreamCallback() { @Override public void process(OutputStream out) throws IOException { out.write("World".getBytes()); } }); contentRepo.disableRead = false; final byte[] buff = new byte["Hello, World".getBytes().length]; session.read(ff, new InputStreamCallback() { @Override public void process(final InputStream in) throws IOException { StreamUtils.fillBuffer(in, buff); } }); assertEquals("Hello, World", new String(buff)); }
@Override public FlowFile append(final FlowFile source, final OutputStreamCallback writer) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); long newSize = 0L; final ContentClaim oldClaim = record.getCurrentClaim(); ByteCountingOutputStream outStream = appendableStreams.get(oldClaim); long originalByteWrittenCount = 0; ContentClaim newClaim = null; try { if (outStream == null) { claimCache.flush(oldClaim); try (final InputStream oldClaimIn = context.getContentRepository().read(oldClaim)) { newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant()); claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source); final OutputStream rawOutStream = context.getContentRepository().write(newClaim); final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream); outStream = new ByteCountingOutputStream(bufferedOutStream); originalByteWrittenCount = 0; appendableStreams.put(newClaim, outStream); StreamUtils.copy(oldClaimIn, outStream); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) { recursionSet.add(source); writer.process(new FlowFileAccessOutputStream(disableOnClose, source)); } finally { recursionSet.remove(source); } } } else { newClaim = oldClaim; originalByteWrittenCount = outStream.getBytesWritten(); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream); final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) { recursionSet.add(source); writer.process(flowFileAccessOutStream); } finally { recursionSet.remove(source); } } newSize = outStream.getBytesWritten(); } catch (final ContentNotFoundException nfe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } handleContentNotFound(nfe, record); } catch (final IOException ioe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe); } catch (final Throwable t) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw t; } finally { if (outStream != null) { final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount; bytesWritten += bytesWrittenThisIteration; } } if (newClaim != oldClaim) { removeTemporaryClaim(record); } final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build(); record.setWorking(newFile); return newFile; }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile append(final FlowFile source, final OutputStreamCallback writer) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); long newSize = 0L; final ContentClaim oldClaim = record.getCurrentClaim(); ByteCountingOutputStream outStream = appendableStreams.get(oldClaim); long originalByteWrittenCount = 0; ContentClaim newClaim = null; try { if (outStream == null) { claimCache.flush(oldClaim); try (final InputStream oldClaimIn = context.getContentRepository().read(oldClaim)) { newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant()); claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source); final OutputStream rawOutStream = context.getContentRepository().write(newClaim); final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream); outStream = new ByteCountingOutputStream(bufferedOutStream); originalByteWrittenCount = 0; appendableStreams.put(newClaim, outStream); StreamUtils.copy(oldClaimIn, outStream); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) { recursionSet.add(source); writer.process(new FlowFileAccessOutputStream(disableOnClose, source)); } finally { recursionSet.remove(source); } } } else { newClaim = oldClaim; originalByteWrittenCount = outStream.getBytesWritten(); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream); final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) { recursionSet.add(source); writer.process(flowFileAccessOutStream); } finally { recursionSet.remove(source); } } newSize = outStream.getBytesWritten(); } catch (final ContentNotFoundException nfe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } handleContentNotFound(nfe, record); } catch (final IOException ioe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe); } catch (final Throwable t) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw t; } finally { if (outStream != null) { final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount; bytesWritten += bytesWrittenThisIteration; } } if (newClaim != oldClaim) { removeTemporaryClaim(record); } final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build(); record.setWorking(newFile); return newFile; } }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile append(final FlowFile source, final OutputStreamCallback writer) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); long newSize = 0L; final ContentClaim oldClaim = record.getCurrentClaim(); ByteCountingOutputStream outStream = appendableStreams.get(oldClaim); long originalByteWrittenCount = 0; ContentClaim newClaim = null; try { if (outStream == null) { claimCache.flush(oldClaim); try (final InputStream oldClaimIn = context.getContentRepository().read(oldClaim)) { newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant()); claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source); final OutputStream rawOutStream = context.getContentRepository().write(newClaim); final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream); outStream = new ByteCountingOutputStream(bufferedOutStream); originalByteWrittenCount = 0; appendableStreams.put(newClaim, outStream); StreamUtils.copy(oldClaimIn, outStream); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) { recursionSet.add(source); writer.process(new FlowFileAccessOutputStream(disableOnClose, source)); } finally { recursionSet.remove(source); } } } else { newClaim = oldClaim; originalByteWrittenCount = outStream.getBytesWritten(); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream); final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) { recursionSet.add(source); writer.process(flowFileAccessOutStream); } finally { recursionSet.remove(source); } } newSize = outStream.getBytesWritten(); } catch (final ContentNotFoundException nfe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } handleContentNotFound(nfe, record); } catch (final IOException ioe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe); } catch (final Throwable t) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw t; } finally { if (outStream != null) { final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount; bytesWritten += bytesWrittenThisIteration; } } if (newClaim != oldClaim) { removeTemporaryClaim(record); } final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build(); record.setWorking(newFile); return newFile; } StandardProcessSession(final ProcessContext context); }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile append(final FlowFile source, final OutputStreamCallback writer) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); long newSize = 0L; final ContentClaim oldClaim = record.getCurrentClaim(); ByteCountingOutputStream outStream = appendableStreams.get(oldClaim); long originalByteWrittenCount = 0; ContentClaim newClaim = null; try { if (outStream == null) { claimCache.flush(oldClaim); try (final InputStream oldClaimIn = context.getContentRepository().read(oldClaim)) { newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant()); claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source); final OutputStream rawOutStream = context.getContentRepository().write(newClaim); final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream); outStream = new ByteCountingOutputStream(bufferedOutStream); originalByteWrittenCount = 0; appendableStreams.put(newClaim, outStream); StreamUtils.copy(oldClaimIn, outStream); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) { recursionSet.add(source); writer.process(new FlowFileAccessOutputStream(disableOnClose, source)); } finally { recursionSet.remove(source); } } } else { newClaim = oldClaim; originalByteWrittenCount = outStream.getBytesWritten(); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream); final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) { recursionSet.add(source); writer.process(flowFileAccessOutStream); } finally { recursionSet.remove(source); } } newSize = outStream.getBytesWritten(); } catch (final ContentNotFoundException nfe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } handleContentNotFound(nfe, record); } catch (final IOException ioe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe); } catch (final Throwable t) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw t; } finally { if (outStream != null) { final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount; bytesWritten += bytesWrittenThisIteration; } } if (newClaim != oldClaim) { removeTemporaryClaim(record); } final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build(); record.setWorking(newFile); return newFile; } StandardProcessSession(final ProcessContext context); void checkpoint(); @Override void commit(); @Override StandardProvenanceEventRecord enrich(final ProvenanceEventRecord rawEvent, final FlowFile flowFile); @Override void rollback(); @Override void rollback(final boolean penalize); @Override void migrate(final ProcessSession newOwner, final Collection<FlowFile> flowFiles); @Override void adjustCounter(final String name, final long delta, final boolean immediate); @Override FlowFile get(); @Override List<FlowFile> get(final int maxResults); @Override List<FlowFile> get(final FlowFileFilter filter); @Override QueueSize getQueueSize(); @Override FlowFile create(); @Override FlowFile clone(final FlowFile example); @Override FlowFile clone(final FlowFile example, final long offset, final long size); @Override FlowFile penalize(final FlowFile flowFile); @Override FlowFile putAttribute(final FlowFile flowFile, final String key, final String value); @Override FlowFile putAllAttributes(final FlowFile flowFile, final Map<String, String> attributes); @Override FlowFile removeAttribute(final FlowFile flowFile, final String key); @Override FlowFile removeAllAttributes(final FlowFile flowFile, final Set<String> keys); @Override FlowFile removeAllAttributes(final FlowFile flowFile, final Pattern keyPattern); @Override void transfer(final FlowFile flowFile, final Relationship relationship); @Override void transfer(final FlowFile flowFile); @Override void transfer(final Collection<FlowFile> flowFiles); @Override void transfer(final Collection<FlowFile> flowFiles, final Relationship relationship); @Override void remove(final FlowFile flowFile); @Override void remove(final Collection<FlowFile> flowFiles); void expireFlowFiles(); @Override void read(final FlowFile source, final InputStreamCallback reader); @Override void read(FlowFile source, boolean allowSessionStreamManagement, InputStreamCallback reader); @Override InputStream read(final FlowFile source); @Override FlowFile merge(final Collection<FlowFile> sources, final FlowFile destination); @Override FlowFile merge(final Collection<FlowFile> sources, final FlowFile destination, final byte[] header, final byte[] footer, final byte[] demarcator); @Override FlowFile write(final FlowFile source, final OutputStreamCallback writer); @Override FlowFile append(final FlowFile source, final OutputStreamCallback writer); @Override FlowFile write(final FlowFile source, final StreamCallback writer); @Override FlowFile importFrom(final Path source, final boolean keepSourceFile, final FlowFile destination); @Override FlowFile importFrom(final InputStream source, final FlowFile destination); @Override void exportTo(final FlowFile source, final Path destination, final boolean append); @Override void exportTo(final FlowFile source, final OutputStream destination); @Override FlowFile create(final FlowFile parent); @Override FlowFile create(final Collection<FlowFile> parents); @Override ProvenanceReporter getProvenanceReporter(); @Override String toString(); }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile append(final FlowFile source, final OutputStreamCallback writer) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); long newSize = 0L; final ContentClaim oldClaim = record.getCurrentClaim(); ByteCountingOutputStream outStream = appendableStreams.get(oldClaim); long originalByteWrittenCount = 0; ContentClaim newClaim = null; try { if (outStream == null) { claimCache.flush(oldClaim); try (final InputStream oldClaimIn = context.getContentRepository().read(oldClaim)) { newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant()); claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source); final OutputStream rawOutStream = context.getContentRepository().write(newClaim); final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream); outStream = new ByteCountingOutputStream(bufferedOutStream); originalByteWrittenCount = 0; appendableStreams.put(newClaim, outStream); StreamUtils.copy(oldClaimIn, outStream); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) { recursionSet.add(source); writer.process(new FlowFileAccessOutputStream(disableOnClose, source)); } finally { recursionSet.remove(source); } } } else { newClaim = oldClaim; originalByteWrittenCount = outStream.getBytesWritten(); try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream); final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) { recursionSet.add(source); writer.process(flowFileAccessOutStream); } finally { recursionSet.remove(source); } } newSize = outStream.getBytesWritten(); } catch (final ContentNotFoundException nfe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } handleContentNotFound(nfe, record); } catch (final IOException ioe) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe); } catch (final Throwable t) { resetWriteClaims(); if (newClaim != oldClaim) { destroyContent(newClaim); } throw t; } finally { if (outStream != null) { final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount; bytesWritten += bytesWrittenThisIteration; } } if (newClaim != oldClaim) { removeTemporaryClaim(record); } final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build(); record.setWorking(newFile); return newFile; } StandardProcessSession(final ProcessContext context); void checkpoint(); @Override void commit(); @Override StandardProvenanceEventRecord enrich(final ProvenanceEventRecord rawEvent, final FlowFile flowFile); @Override void rollback(); @Override void rollback(final boolean penalize); @Override void migrate(final ProcessSession newOwner, final Collection<FlowFile> flowFiles); @Override void adjustCounter(final String name, final long delta, final boolean immediate); @Override FlowFile get(); @Override List<FlowFile> get(final int maxResults); @Override List<FlowFile> get(final FlowFileFilter filter); @Override QueueSize getQueueSize(); @Override FlowFile create(); @Override FlowFile clone(final FlowFile example); @Override FlowFile clone(final FlowFile example, final long offset, final long size); @Override FlowFile penalize(final FlowFile flowFile); @Override FlowFile putAttribute(final FlowFile flowFile, final String key, final String value); @Override FlowFile putAllAttributes(final FlowFile flowFile, final Map<String, String> attributes); @Override FlowFile removeAttribute(final FlowFile flowFile, final String key); @Override FlowFile removeAllAttributes(final FlowFile flowFile, final Set<String> keys); @Override FlowFile removeAllAttributes(final FlowFile flowFile, final Pattern keyPattern); @Override void transfer(final FlowFile flowFile, final Relationship relationship); @Override void transfer(final FlowFile flowFile); @Override void transfer(final Collection<FlowFile> flowFiles); @Override void transfer(final Collection<FlowFile> flowFiles, final Relationship relationship); @Override void remove(final FlowFile flowFile); @Override void remove(final Collection<FlowFile> flowFiles); void expireFlowFiles(); @Override void read(final FlowFile source, final InputStreamCallback reader); @Override void read(FlowFile source, boolean allowSessionStreamManagement, InputStreamCallback reader); @Override InputStream read(final FlowFile source); @Override FlowFile merge(final Collection<FlowFile> sources, final FlowFile destination); @Override FlowFile merge(final Collection<FlowFile> sources, final FlowFile destination, final byte[] header, final byte[] footer, final byte[] demarcator); @Override FlowFile write(final FlowFile source, final OutputStreamCallback writer); @Override FlowFile append(final FlowFile source, final OutputStreamCallback writer); @Override FlowFile write(final FlowFile source, final StreamCallback writer); @Override FlowFile importFrom(final Path source, final boolean keepSourceFile, final FlowFile destination); @Override FlowFile importFrom(final InputStream source, final FlowFile destination); @Override void exportTo(final FlowFile source, final Path destination, final boolean append); @Override void exportTo(final FlowFile source, final OutputStream destination); @Override FlowFile create(final FlowFile parent); @Override FlowFile create(final Collection<FlowFile> parents); @Override ProvenanceReporter getProvenanceReporter(); @Override String toString(); static final int VERBOSE_LOG_THRESHOLD; static final String DEFAULT_FLOWFILE_PATH; }