D-Net

package eu.dnetlib.msro.workflows.nodes.functions;

import java.util.function.Function;

import java.util.stream.Stream;

import eu.dnetlib.msro.exceptions.MSROException;

import eu.dnetlib.msro.workflows.Arc;

import eu.dnetlib.msro.workflows.nodes.SimpleProcessNode;

import eu.dnetlib.msro.workflows.procs.Token;

public abstract class AbstractApplyFunctionJobNode<T, K> extends SimpleProcessNode {

	private Stream<T> inputStream;

	private Stream<K> outputStream;

	@Override

	protected final String execute(final Token token) throws Exception {

		if (inputStream == null) { throw new MSROException("inputStream not found in ENV"); }

		outputStream = inputStream.map(getFunction());

		return Arc.DEFAULT_ARC;

	}

	protected abstract Function<T, K> getFunction() throws MSROException;

	public Stream<T> getInputStream() {

		return inputStream;

	}

	public void setInputStream(final Stream<T> inputStream) {

		this.inputStream = inputStream;

	}

	public Stream<K> getOutputStream() {

		return outputStream;

	}

	public void setOutputStream(final Stream<K> outputStream) {

		this.outputStream = outputStream;

	}

}