D-Net

package eu.dnetlib.openaire.exporter;

import java.io.IOException;

import java.io.OutputStream;

import java.nio.charset.Charset;

import java.sql.*;

import java.time.Duration;

import java.time.LocalDateTime;

import java.util.Arrays;

import java.util.List;

import java.util.zip.GZIPOutputStream;

import java.util.zip.ZipOutputStream;

import com.google.common.base.Function;

import com.google.common.base.Joiner;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.openaire.exporter.tmpModel.Project;

import eu.dnetlib.openaire.exporter.tmpModel.ProjectDetail;

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.beans.factory.annotation.Value;

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

	@Value("${openaire.exporter.jdbc.maxrows}")

	private int maxRows;

	@Value("${openaire.exporter.projectdetails.flushsize}")

	private int gzipFlushSize;

	@Value("${openaire.exporter.projects2tsv.fields}")

	private String tsvFields;

	@Autowired

	private BasicDataSource apiDataSource;

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

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

			final ValueCleaner cleaner) throws IOException, SQLException {

		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();

		}

	}

	public void streamProjectsTSV(final String sql, final ZipOutputStream out) throws IOException, SQLException {

		if (log.isDebugEnabled()) {

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

		}

		final LocalDateTime start = LocalDateTime.now();

		final List<String> fields = Lists.newArrayList(Splitter.on(",").omitEmptyStrings().trimResults().split(tsvFields));

		writeCSVLine(out, fields);

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

			while(rs.next()) {

				final Project p = new Project()

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

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

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

						.setCallIdentifier(rs.getString("call_identifier"))

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

						.setEnddate(rs.getString("enddate"))

						.setOaMandateForPublications(rs.getBoolean("oa_mandate_for_publications"))

						.setOaMandateForDatasets(rs.getBoolean("oa_mandate_for_datasets"))

						.setDescription(rs.getString("description"))

						.setOrgLegalname(rs.getString("legalname"))

						.setOrgCountry(rs.getString("country"))

						.setOrgRole(rs.getString("role"))

						.setFirstname(rs.getString("firstname"))

						.setSecondnames(rs.getString("secondnames"))

						.setEmail(rs.getString("email"));

				writeCSVLine(out, p.asList());

			}

			out.closeEntry();

			final LocalDateTime end = LocalDateTime.now();

			if (log.isDebugEnabled()) {

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

			}

		} finally {

			out.close();

		}

	}

	private void writeCSVLine(final ZipOutputStream out, final List<String> list) throws IOException {

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

		out.write('\n');

	}

	public void streamProjectDetails(final String sql, final OutputStream out, final String format) throws SQLException, IOException {

		if (log.isDebugEnabled()) {

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

		}

		final LocalDateTime start = LocalDateTime.now();

		int i = 0;

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

			while (rs.next()) {

				final ProjectDetail p = getProjectDetail(rs);

				switch (format) {

				case "csv":

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

					break;

				case "json":

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

					break;

				}

				if (++i % gzipFlushSize == 0) {

					log.debug("flushing output stream");

					out.flush();

				}

			}

			final LocalDateTime end = LocalDateTime.now();

			if (log.isDebugEnabled()) {

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

			}

		} finally {

			if (out instanceof GZIPOutputStream) {

				((GZIPOutputStream) out).finish();

			}

			out.close();

		}

	}

	private Connection getConn() throws SQLException {

		final Connection conn = apiDataSource.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(maxRows);

		return stm;

	}

	private ProjectDetail getProjectDetail(final ResultSet rs) throws SQLException {

		return new ProjectDetail()

				.setProjectId(rs.getString("projectid"))

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

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

				.setJsonextrainfo(rs.getString("jsonextrainfo"))

				.setFundingPath(asList(rs.getArray("fundingpath")));

	}

	private List<String> asList(final Array value) throws SQLException {

		if (value != null) {

			final List<Object> list = Arrays.asList((Object[]) value.getArray());

			return Lists.newArrayList(Iterables.transform(list, new Function<Object, String>() {

				@Override

				public String apply(final Object o) {

					return o != null ? o.toString() : null;

				}

			}));

		}

		return Lists.newArrayList();

	}

}