@Test public void testExportTo() throws IOException { final ContentClaim claim = contentRepo.create(false); final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder() .contentClaim(claim) .addAttribute("uuid", "12345678-1234-1234-1234-123456789012") .entryDate(System.currentTimeMillis()) .build(); flowFileQueue.put(flowFileRecord); FlowFile flowFile = session.get(); assertNotNull(flowFile); flowFile = session.append(flowFile, new OutputStreamCallback() { @Override public void process(OutputStream out) throws IOException { out.write("Hello World".getBytes()); } }); ByteArrayOutputStream os = new ByteArrayOutputStream(); session.exportTo(flowFile, os); assertEquals("Hello World", new String(os.toByteArray())); os.close(); FileOutputStream mock = Mockito.mock(FileOutputStream.class); doThrow(new IOException()).when(mock).write((byte[]) notNull(), any(Integer.class), any(Integer.class)); try { session.exportTo(flowFile, mock); Assert.fail("Expected ProcessException"); } catch (ProcessException e) { } }
@Override public void exportTo(final FlowFile source, final Path destination, final boolean append) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); try { ensureNotAppending(record.getCurrentClaim()); claimCache.flush(record.getCurrentClaim()); final long copyCount = context.getContentRepository().exportTo(record.getCurrentClaim(), destination, append, record.getCurrentClaimOffset(), source.getSize()); bytesRead += copyCount; bytesWritten += copyCount; } catch (final ContentNotFoundException nfe) { handleContentNotFound(nfe, record); } catch (final Throwable t) { throw new FlowFileAccessException("Failed to export " + source + " to " + destination + " due to " + t.toString(), t); } }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public void exportTo(final FlowFile source, final Path destination, final boolean append) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); try { ensureNotAppending(record.getCurrentClaim()); claimCache.flush(record.getCurrentClaim()); final long copyCount = context.getContentRepository().exportTo(record.getCurrentClaim(), destination, append, record.getCurrentClaimOffset(), source.getSize()); bytesRead += copyCount; bytesWritten += copyCount; } catch (final ContentNotFoundException nfe) { handleContentNotFound(nfe, record); } catch (final Throwable t) { throw new FlowFileAccessException("Failed to export " + source + " to " + destination + " due to " + t.toString(), t); } } }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public void exportTo(final FlowFile source, final Path destination, final boolean append) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); try { ensureNotAppending(record.getCurrentClaim()); claimCache.flush(record.getCurrentClaim()); final long copyCount = context.getContentRepository().exportTo(record.getCurrentClaim(), destination, append, record.getCurrentClaimOffset(), source.getSize()); bytesRead += copyCount; bytesWritten += copyCount; } catch (final ContentNotFoundException nfe) { handleContentNotFound(nfe, record); } catch (final Throwable t) { throw new FlowFileAccessException("Failed to export " + source + " to " + destination + " due to " + t.toString(), t); } } StandardProcessSession(final ProcessContext context); }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public void exportTo(final FlowFile source, final Path destination, final boolean append) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); try { ensureNotAppending(record.getCurrentClaim()); claimCache.flush(record.getCurrentClaim()); final long copyCount = context.getContentRepository().exportTo(record.getCurrentClaim(), destination, append, record.getCurrentClaimOffset(), source.getSize()); bytesRead += copyCount; bytesWritten += copyCount; } catch (final ContentNotFoundException nfe) { handleContentNotFound(nfe, record); } catch (final Throwable t) { throw new FlowFileAccessException("Failed to export " + source + " to " + destination + " due to " + t.toString(), t); } } 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 void exportTo(final FlowFile source, final Path destination, final boolean append) { validateRecordState(source); final StandardRepositoryRecord record = records.get(source); try { ensureNotAppending(record.getCurrentClaim()); claimCache.flush(record.getCurrentClaim()); final long copyCount = context.getContentRepository().exportTo(record.getCurrentClaim(), destination, append, record.getCurrentClaimOffset(), source.getSize()); bytesRead += copyCount; bytesWritten += copyCount; } catch (final ContentNotFoundException nfe) { handleContentNotFound(nfe, record); } catch (final Throwable t) { throw new FlowFileAccessException("Failed to export " + source + " to " + destination + " due to " + t.toString(), t); } } 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; }