@Test public void testGetWithCount() { for (int i = 0; i < 8; i++) { final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder() .id(i) .addAttribute("uuid", "000000000000-0000-0000-0000-0000000" + i) .build(); this.flowFileQueue.put(flowFile); } final List<FlowFile> flowFiles = session.get(7); assertEquals(7, flowFiles.size()); }
@Override public FlowFile get() { final List<Connection> connections = context.getPollableConnections(); final int numConnections = connections.size(); for (int numAttempts = 0; numAttempts < numConnections; numAttempts++) { final Connection conn = connections.get(context.getNextIncomingConnectionIndex() % numConnections); final Set<FlowFileRecord> expired = new HashSet<>(); final FlowFileRecord flowFile = conn.getFlowFileQueue().poll(expired); removeExpired(expired, conn); if (flowFile != null) { registerDequeuedRecord(flowFile, conn); return flowFile; } } return null; }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile get() { final List<Connection> connections = context.getPollableConnections(); final int numConnections = connections.size(); for (int numAttempts = 0; numAttempts < numConnections; numAttempts++) { final Connection conn = connections.get(context.getNextIncomingConnectionIndex() % numConnections); final Set<FlowFileRecord> expired = new HashSet<>(); final FlowFileRecord flowFile = conn.getFlowFileQueue().poll(expired); removeExpired(expired, conn); if (flowFile != null) { registerDequeuedRecord(flowFile, conn); return flowFile; } } return null; } }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile get() { final List<Connection> connections = context.getPollableConnections(); final int numConnections = connections.size(); for (int numAttempts = 0; numAttempts < numConnections; numAttempts++) { final Connection conn = connections.get(context.getNextIncomingConnectionIndex() % numConnections); final Set<FlowFileRecord> expired = new HashSet<>(); final FlowFileRecord flowFile = conn.getFlowFileQueue().poll(expired); removeExpired(expired, conn); if (flowFile != null) { registerDequeuedRecord(flowFile, conn); return flowFile; } } return null; } StandardProcessSession(final ProcessContext context); }
StandardProcessSession implements ProcessSession, ProvenanceEventEnricher { @Override public FlowFile get() { final List<Connection> connections = context.getPollableConnections(); final int numConnections = connections.size(); for (int numAttempts = 0; numAttempts < numConnections; numAttempts++) { final Connection conn = connections.get(context.getNextIncomingConnectionIndex() % numConnections); final Set<FlowFileRecord> expired = new HashSet<>(); final FlowFileRecord flowFile = conn.getFlowFileQueue().poll(expired); removeExpired(expired, conn); if (flowFile != null) { registerDequeuedRecord(flowFile, conn); return flowFile; } } return null; } 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 get() { final List<Connection> connections = context.getPollableConnections(); final int numConnections = connections.size(); for (int numAttempts = 0; numAttempts < numConnections; numAttempts++) { final Connection conn = connections.get(context.getNextIncomingConnectionIndex() % numConnections); final Set<FlowFileRecord> expired = new HashSet<>(); final FlowFileRecord flowFile = conn.getFlowFileQueue().poll(expired); removeExpired(expired, conn); if (flowFile != null) { registerDequeuedRecord(flowFile, conn); return flowFile; } } return null; } 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; }