package eu.dnetlib.manager.wf.nodes.stream;

import java.util.stream.Stream;

import eu.dnetlib.manager.wf.annotations.WfInputParam;
import eu.dnetlib.manager.wf.annotations.WfOutputParam;
import eu.dnetlib.manager.wf.nodes.AbstractJobNode;
import eu.dnetlib.manager.wf.workflows.procs.Token;

public abstract class StreamFilterNode<T> extends AbstractJobNode {

	@WfInputParam
	private Stream<T> inputStream;

	@WfOutputParam
	private Stream<T> outputStream;

	public StreamFilterNode() {
		super(false);
	}

	abstract protected Stream<T> filterStream(Stream<T> input, Token token);

	@Override
	protected final void execute(final Token token) {
		outputStream = filterStream(inputStream, token);
	}

}