D-Net

package eu.dnetlib.data.transform.xml.vtd;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import eu.dnetlib.data.transform.xml2.DatasetToProto;

import org.apache.commons.lang3.time.StopWatch;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;

import org.bson.Document;

import org.junit.Test;

import org.junit.runner.RunWith;

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

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.IOException;

import java.util.Objects;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.stream.StreamSupport;

import static org.junit.Assert.assertNotNull;

import static org.junit.Assert.assertTrue;

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(classes = { ConfigurationTestConfig.class })

public class VtdParserToProtoIT {

	private static final Log log = LogFactory.getLog(VtdParserToProtoIT.class);

	public static final String COLLECTION_NAME = "datacite";

	private static final int BATCH_SIZE = 10000;

	public static final int LOG_FREQ = 5000;

	@Autowired

	private MongoDatabase db;

	@Test

	public void testParseDatacite() throws IOException {

		final MongoCollection<Document> collection = db.getCollection(COLLECTION_NAME);

		final long collectionSize = collection.count();

		log.info(String.format("found %s records in collection '%s'", collectionSize, COLLECTION_NAME));

		final AtomicInteger read = new AtomicInteger(0);

		final DescriptiveStatistics stats = new DescriptiveStatistics();

		final StopWatch timer = new StopWatch();

		final DatasetToProto mapper = new DatasetToProto();

		StreamSupport.stream(collection.find().batchSize(BATCH_SIZE).spliterator(), false)

				.peek(d -> {

					if (read.addAndGet(1) % LOG_FREQ == 0) {

						log.info(String.format("records read so far %s", read.get()));

						log.info(String.format("stats so far %s", stats.toString()));

					}

				})

				.map(d -> (String) d.get("body"))

				.filter(Objects::nonNull)

				.peek(s -> timer.start())

				.map(mapper)

				.forEach(oaf -> {

					assertNotNull(oaf);

					assertTrue(oaf.hasEntity());

					timer.stop();

					stats.addValue(timer.getTime());

					timer.reset();

				});

		log.info(String.format("processed %s/%s records", read.get(), collectionSize));

	}

}