@Test public void testProcess() throws InterruptedException, LifecycleException, EventDeliveryException { Channel channel = new PseudoTxnMemoryChannel(); Context context = new Context(); context.put("logicalNode.name", "test"); Configurables.configure(source, context); Configurables.configure(channel, context); List<Channel> channels = new ArrayList<Channel>(); channels.add(channel); ChannelSelector rcs = new ReplicatingChannelSelector(); rcs.setChannels(channels); source.setChannelProcessor(new ChannelProcessor(rcs)); for (long i = 0; i < 100; i++) { source.process(); Event event = channel.take(); Assert.assertArrayEquals(String.valueOf(i).getBytes(), new String(event.getBody()).getBytes()); } }
@Override public Status process() throws EventDeliveryException { Status status = Status.READY; int i = 0; try { if (batchSize <= 1) { if(eventsSent < totalEvents) { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); sourceCounter.incrementEventAcceptedCount(); eventsSent++; } else { status = Status.BACKOFF; } } else { batchArrayList.clear(); for (i = 0; i < batchSize; i++) { if(eventsSent < totalEvents){ batchArrayList.add(i, EventBuilder.withBody(String .valueOf(sequence++).getBytes())); eventsSent++; } else { status = Status.BACKOFF; } } if(!batchArrayList.isEmpty()) { getChannelProcessor().processEventBatch(batchArrayList); sourceCounter.incrementAppendBatchAcceptedCount(); sourceCounter.addToEventAcceptedCount(batchArrayList.size()); } } } catch (ChannelException ex) { eventsSent -= i; logger.error( getName() + " source could not write to channel.", ex); } return status; }
SequenceGeneratorSource extends AbstractSource implements
    PollableSource, Configurable { @Override public Status process() throws EventDeliveryException { Status status = Status.READY; int i = 0; try { if (batchSize <= 1) { if(eventsSent < totalEvents) { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); sourceCounter.incrementEventAcceptedCount(); eventsSent++; } else { status = Status.BACKOFF; } } else { batchArrayList.clear(); for (i = 0; i < batchSize; i++) { if(eventsSent < totalEvents){ batchArrayList.add(i, EventBuilder.withBody(String .valueOf(sequence++).getBytes())); eventsSent++; } else { status = Status.BACKOFF; } } if(!batchArrayList.isEmpty()) { getChannelProcessor().processEventBatch(batchArrayList); sourceCounter.incrementAppendBatchAcceptedCount(); sourceCounter.addToEventAcceptedCount(batchArrayList.size()); } } } catch (ChannelException ex) { eventsSent -= i; logger.error( getName() + " source could not write to channel.", ex); } return status; } }
SequenceGeneratorSource extends AbstractSource implements
    PollableSource, Configurable { @Override public Status process() throws EventDeliveryException { Status status = Status.READY; int i = 0; try { if (batchSize <= 1) { if(eventsSent < totalEvents) { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); sourceCounter.incrementEventAcceptedCount(); eventsSent++; } else { status = Status.BACKOFF; } } else { batchArrayList.clear(); for (i = 0; i < batchSize; i++) { if(eventsSent < totalEvents){ batchArrayList.add(i, EventBuilder.withBody(String .valueOf(sequence++).getBytes())); eventsSent++; } else { status = Status.BACKOFF; } } if(!batchArrayList.isEmpty()) { getChannelProcessor().processEventBatch(batchArrayList); sourceCounter.incrementAppendBatchAcceptedCount(); sourceCounter.addToEventAcceptedCount(batchArrayList.size()); } } } catch (ChannelException ex) { eventsSent -= i; logger.error( getName() + " source could not write to channel.", ex); } return status; } SequenceGeneratorSource(); }
SequenceGeneratorSource extends AbstractSource implements
    PollableSource, Configurable { @Override public Status process() throws EventDeliveryException { Status status = Status.READY; int i = 0; try { if (batchSize <= 1) { if(eventsSent < totalEvents) { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); sourceCounter.incrementEventAcceptedCount(); eventsSent++; } else { status = Status.BACKOFF; } } else { batchArrayList.clear(); for (i = 0; i < batchSize; i++) { if(eventsSent < totalEvents){ batchArrayList.add(i, EventBuilder.withBody(String .valueOf(sequence++).getBytes())); eventsSent++; } else { status = Status.BACKOFF; } } if(!batchArrayList.isEmpty()) { getChannelProcessor().processEventBatch(batchArrayList); sourceCounter.incrementAppendBatchAcceptedCount(); sourceCounter.addToEventAcceptedCount(batchArrayList.size()); } } } catch (ChannelException ex) { eventsSent -= i; logger.error( getName() + " source could not write to channel.", ex); } return status; } SequenceGeneratorSource(); @Override void configure(Context context); @Override Status process(); @Override void start(); @Override void stop(); }
SequenceGeneratorSource extends AbstractSource implements
    PollableSource, Configurable { @Override public Status process() throws EventDeliveryException { Status status = Status.READY; int i = 0; try { if (batchSize <= 1) { if(eventsSent < totalEvents) { getChannelProcessor().processEvent( EventBuilder.withBody(String.valueOf(sequence++).getBytes())); sourceCounter.incrementEventAcceptedCount(); eventsSent++; } else { status = Status.BACKOFF; } } else { batchArrayList.clear(); for (i = 0; i < batchSize; i++) { if(eventsSent < totalEvents){ batchArrayList.add(i, EventBuilder.withBody(String .valueOf(sequence++).getBytes())); eventsSent++; } else { status = Status.BACKOFF; } } if(!batchArrayList.isEmpty()) { getChannelProcessor().processEventBatch(batchArrayList); sourceCounter.incrementAppendBatchAcceptedCount(); sourceCounter.addToEventAcceptedCount(batchArrayList.size()); } } } catch (ChannelException ex) { eventsSent -= i; logger.error( getName() + " source could not write to channel.", ex); } return status; } SequenceGeneratorSource(); @Override void configure(Context context); @Override Status process(); @Override void start(); @Override void stop();  }