D-Net

package eu.dnetlib.msro.workers.aggregation.collect;

import java.io.StringReader;

import java.util.stream.Stream;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.msro.annotations.Direction;

import eu.dnetlib.msro.annotations.EnvParam;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workers.aggregation.objects.InterfaceDescriptor;

import eu.dnetlib.msro.workflows.graph.Arc;

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

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

@Component

@Scope("prototype")

@ProcessNode("Collect")

public class CollectJobNode extends AbstractProcessNode {

	@Autowired

	private InformationServiceClient isLookup;

	@Autowired

	private MetadataCollector collector;

	private String datasourceId;

	private String datasourceInterface;

	@EnvParam(value = "eprParam", direction = Direction.OUT, reference = true)

	private Stream<String> outputStream;

	@Override

	protected String execute() throws Exception {

		final String profile = isLookup.getProfile(datasourceId);

		final Document doc = new SAXReader().read(new StringReader(profile));

		final Node ifcNode = doc.selectSingleNode("//INTERFACE[@id='" + datasourceInterface + "']");

		final InterfaceDescriptor interfaceDescriptor = InterfaceDescriptor.newInstance(ifcNode);

		outputStream = collector.collect(interfaceDescriptor);

		return Arc.DEFAULT_ARC;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public Stream<String> getOutputStream() {

		return outputStream;

	}

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

		this.outputStream = outputStream;

	}

}