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;
	}

}