D-Net

package eu.dnetlib.data.transform;

import java.util.List;

import java.util.function.Function;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.mapreduce.util.OafEntityDecoder;

import eu.dnetlib.data.proto.OafProtos.Oaf;

public class OafToRowMapper implements Function<Oaf, List<Row>> {

	public static final String BODY = "body";

	@Override

	public List<Row> apply(final Oaf oaf) {

		final List<Row> rows = Lists.newArrayList();

		final OafDecoder d = OafDecoder.decode(oaf);

		final OafEntityDecoder entity = d.decodeEntity();

		final Row r = new Row(d.getCFQ(), entity.getId());

		switch (entity.getType()) {

		case project:

			r.addColumn(new Column<>(BODY, oaf.toByteArray()));

			break;

		case result:

			oaf.getEntity().getCachedOafRelList().stream()

					.map(cachedRel -> {

						final Oaf.Builder oafRel = Oaf.newBuilder(cachedRel);

						oafRel.getRelBuilder().clearCachedOafTarget();

						return oafRel.build();

					}).forEach(oafRel -> r.addColumn(new Column<>(OafDecoder.decode(oafRel).getCFQ(), oafRel.toByteArray())));

			break;

		case datasource:

			break;

		case organization:

			break;

		}

		return rows;

	}

}