D-Net

package eu.dnetlib.openaire.exporter.project;

import java.io.BufferedOutputStream;

import java.io.IOException;

import java.io.OutputStream;

import java.nio.charset.Charset;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.time.Duration;

import java.time.LocalDateTime;

import java.util.List;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.zip.GZIPOutputStream;

import java.util.zip.ZipEntry;

import java.util.zip.ZipOutputStream;

import javax.annotation.PostConstruct;

import com.google.common.base.Joiner;

import com.google.common.base.Splitter;

import com.google.common.util.concurrent.ListeningExecutorService;

import com.google.common.util.concurrent.MoreExecutors;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import eu.dnetlib.OpenaireExporterConfig;

import eu.dnetlib.openaire.exporter.datasource.ApiException;

import eu.dnetlib.openaire.exporter.model.project.Project;

import eu.dnetlib.openaire.exporter.model.project.ProjectTsv;

import eu.dnetlib.openaire.exporter.project.repository.ProjectDetailsRepository;

import eu.dnetlib.openaire.exporter.project.repository.ProjectTsvRepository;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.dbcp2.BasicDataSource;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 20/09/16.

 */

@Component

public class JdbcApiDao {

	public static final Charset UTF8 = Charset.forName("UTF-8");

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

	@Autowired

	private OpenaireExporterConfig config;

	@Autowired

	private BasicDataSource projectApiDataSource;

	@Autowired

	private ProjectDetailsRepository projectDetailsRepository;

	@Autowired

	private ProjectTsvRepository projectTsvRepository;

	private ListeningExecutorService service;

	@PostConstruct

	public void init() {

		service = MoreExecutors.listeningDecorator(

				new ScheduledThreadPoolExecutor(config.getRequestWorkers(),

						new ThreadFactoryBuilder().setNameFormat("project-exporter-dao-%d").build()));

	}

	public void processProjectDetails(final OutputStream outputStream, String format, Boolean compress) throws IOException {

		service.execute(() -> {

			try {

				final OutputStream out = getOutputStream(outputStream, compress);

				try {

					final AtomicInteger i = new AtomicInteger(0);

					projectDetailsRepository.findAll().forEach(p -> {

							try {

								switch (format) {

								case "csv":

									out.write(p.asCSV().getBytes(UTF8));

									break;

								case "json":

									out.write(p.asJson().getBytes(UTF8));

									break;

								}

								i.incrementAndGet();

								if (i.intValue() % config.getProject().getGzipFlushSize() == 0) {

									log.debug("flushing output stream");

									out.flush();

								}

							} catch (IOException e) {

								throw new RuntimeException(e);

							}

						}

					);

				} finally {

					if (out instanceof GZIPOutputStream) {

						((GZIPOutputStream) out).finish();

					}

					out.close();

				}

			} catch (IOException e) {

				throw new RuntimeException(e);

			}

		});

	}

	private OutputStream getOutputStream(final OutputStream outputStream, final Boolean compress) throws IOException {

		if (compress != null && compress) {

			return new GZIPOutputStream(outputStream);

		}

		return outputStream;

	}

	public void processTsvRequest(final OutputStream outputStream, final Boolean article293, final String fundingPrefix, final String filename) throws ApiException {

		service.execute(() -> {

			try(final ZipOutputStream out = new ZipOutputStream(new BufferedOutputStream(outputStream))) {

				out.putNextEntry(new ZipEntry(filename));

				writeTsvLine(out, Splitter.on(",").trimResults().splitToList(config.getProject().getTsvFields()));

				queryForTsv(fundingPrefix, article293).forEach(p -> {

					try {

						writeTsvLine(out, p.asList());

					} catch (IOException e) {

						throw new RuntimeException(e);

					}

				});

			} catch (Throwable e) {

				throw new RuntimeException("Error processing the request", e);

			}

		});

	}

	private void writeTsvLine(final ZipOutputStream out, final List<String> s) throws IOException {

		out.write(Joiner.on('\t').useForNull("").join(s).getBytes(UTF8));

		out.write('\n');

	}

	private Iterable<ProjectTsv> queryForTsv(final String fundingPrefix, final Boolean article293) {

		log.debug(String.format("fundingPrefix:'%s' and article293:'%s'", fundingPrefix, article293));

		if (article293 != null) {

			return projectTsvRepository.findByFundingpathidStartingWithAndOaMandateForDatasetsOrderByAcronym(fundingPrefix, article293);

		} else {

			return projectTsvRepository.findByFundingpathidStartingWithOrderByAcronym(fundingPrefix);

		}

	}

	public void streamProjects(final String sql, final OutputStream out,

			final String head, final StringTemplate projectTemplate, final String tail,

			final ValueCleaner cleaner) throws IOException, SQLException {

		service.execute(() -> {

			try {

				if (log.isDebugEnabled()) {

					log.debug("Thread " + Thread.currentThread().getId() + " begin");

				}

				final LocalDateTime start = LocalDateTime.now();

				if (StringUtils.isNotBlank(head)) {

					out.write(head.getBytes(UTF8));

				}

				try (final Connection con = getConn(); final PreparedStatement stm = getStm(sql, con); final ResultSet rs = stm.executeQuery()) {

					while (rs.next()) {

						final Project p = new Project()

								.setFunder(cleaner.clean(rs.getString("funder")))

								.setJurisdiction(cleaner.clean(rs.getString("jurisdiction")))

								.setFundingpathid(cleaner.clean(rs.getString("fundingpathid")))

								.setAcronym(cleaner.clean(rs.getString("acronym")))

								.setTitle(cleaner.clean(rs.getString("title")))

								.setCode(cleaner.clean(rs.getString("code")))

								.setStartdate(cleaner.clean(rs.getString("startdate")))

								.setEnddate(cleaner.clean(rs.getString("enddate")));

						projectTemplate.reset();

						projectTemplate.setAttribute("p", p);

						out.write(projectTemplate.toString().getBytes(UTF8));

					}

					if (StringUtils.isNotBlank(tail)) {

						out.write(tail.getBytes(UTF8));

					}

					final LocalDateTime end = LocalDateTime.now();

					if (log.isDebugEnabled()) {

						log.debug("Thread " + Thread.currentThread().getId() + " ends, took: " + Duration.between(start, end));

					}

				} finally {

					out.close();

				}

			} catch (IOException | SQLException e) {

				log.error(e);

				throw new RuntimeException(e);

			}

		});

	}

	private Connection getConn() throws SQLException {

		final Connection conn = projectApiDataSource.getConnection();

		conn.setAutoCommit(false);

		return conn;

	}

	private PreparedStatement getStm(final String sql, final Connection con) throws SQLException {

		final PreparedStatement stm = con.prepareStatement(sql);

		stm.setFetchSize(config.getJdbc().getMaxRows());

		return stm;

	}

}