The following code shows a simplified producer stage taken from the original InitialElementProducer. It takes a list of elements and sends them one by one to its output port.

public class InitialElementProducer<T> extends AbstractProducerStage<T> {

	private final Collection<T> elements;

	public InitialElementProducer(final Collection<T> elements) {
		this.elements = elements;
	}

	@Override
	public void onStarting() throws Exception {
		if (elements == null) {
			throw new IllegalArgumentException("The given iterable must not be null");
		}
		super.onStarting();
	}

	@Override
	protected void execute() {
		for (final T element : this.elements) {
			this.outputPort.send(element);
		}
		this.workCompleted();
	}

}

First, we implement execute. It simply iterates over the given list and sends all elements by calling this.outputPort.send(element).

Furthermore, we want to check if the given iterable is not null. This can be achieved by running a check while starting the stage. For this, we override the method onStarting and add a null check to it. This method will be called once by the framework before it executes the stage for the first time. Note: You also need to make sure that the super method is called, too.

Finally, we need to tell the framework that this stage has finished its work. Calling the method workCompleted() will do so. Moreover, it automatically propagates a termination signal through the whole P&F architecture. In this way, all stages terminate themselves after processing their inputs. Thus, no external termination action is required by the user.

Have a look at the Clock stage, as an additional example, which outputs the current timestamp as long as a certain condition holds.