D-Net

package eu.dnetlib.enabling.is.sn;

import java.io.IOException;

import java.sql.Connection;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.Statement;

import java.util.HashMap;

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.cxf.helpers.IOUtils;

import org.postgresql.PGConnection;

import org.postgresql.PGNotification;

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

import com.google.gson.Gson;

import com.google.gson.reflect.TypeToken;

import eu.dnetlib.enabling.is.jdbc.DatabaseUtils;

import eu.dnetlib.enabling.is.rmi.Operation;

public class NotificationListener {

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

	private DatabaseUtils dbUtils;

	private NotificationDispatcher dispatcher;

	private boolean running = false;

	private Connection conn = null;

	private String query = "";

	private Gson gson = new Gson();

	public void init() throws IOException {

		try {

			this.query = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/enabling/is/templates/obtain_notifications.sql.st"));

			this.conn = dbUtils.getDataSource().getConnection();

			startListening("insert", "update", "delete");

		} catch (SQLException e) {

			log.error("Error opening connection", e);

		}

	}

	public void destroy() {

		try {

			if (conn != null && !conn.isClosed()) {

				conn.close();

			}

		} catch (SQLException e) {

			log.error("Error closing connection", e);

		}

	}

	public void verify() {

		if (running) { return; }

		synchronized (this) {

			running = true;

			try {

				final Statement stmt = conn.createStatement();

				final ResultSet rs = stmt.executeQuery("SELECT 1");

				rs.close();

				stmt.close();

				final PGNotification notifications[] = ((PGConnection) conn).getNotifications();

				if (notifications != null && notifications.length > 0) {

					if (log.isInfoEnabled()) {

						for (PGNotification notification : notifications) {

							log.info("Notification " + notification.getName() + " on table " + notification.getParameter());

						}

					}

					for (Map<String, Object> map : dbUtils.searchSimple(query)) {

						final Operation operation = Operation.valueOf((String) map.get("operation"));

						final Map<String, Object> mapCond = obtainMapFromJson((String) map.get("condition"));

						final Map<String, Object> mapTest = obtainMapFromJson((String) map.get(operation == Operation.DELETE ? "old" : "new"));

						if (compareMaps(mapCond, mapTest)) {

							log.info("SENDING NOTIFICATIN TO " + map.get("service_id"));

							final NotificationMessage message = new NotificationMessage();

							message.setSubscriptionId((String) map.get("subscr_id"));

							message.setIsId("XXXX");

							message.setMessage((String) map.get(operation == Operation.DELETE ? "old" : "new"));

							message.setServiceId((String) map.get("service_id"));

							message.setTopic(map.get("operation") + "/" + map.get("tbl"));

							dispatcher.sendNotification(message);

						}

					}

				}

			} catch (Exception e) {

				log.error(e);

			}

			running = false;

		}

	}

	private Map<String, Object> obtainMapFromJson(final String s) {

		if (StringUtils.isBlank(s)) {

			return new HashMap<String, Object>();

		} else {

			return gson.fromJson(s, new TypeToken<Map<String, Object>>() {}.getType());

		}

	}

	private boolean compareMaps(final Map<String, Object> cond, final Map<String, Object> map) {

		if (cond != null) {

			for (Entry<String, Object> ec : cond.entrySet()) {

				final Object v1 = ec.getValue();

				final Object v2 = map.get(ec.getKey());

				if (v1 != null && v2 != null && !v1.equals(v2)) { return false; }

				if (v1 == null && v2 != null) { return false; }

				if (v1 != null && v2 == null) { return false; }

			}

		}

		return true;

	}

	private void startListening(final String... messages) throws SQLException {

		for (String message : messages) {

			final Statement stmt = conn.createStatement();

			stmt.execute("LISTEN " + message);

			stmt.close();

		}

	}

	public DatabaseUtils getDbUtils() {

		return dbUtils;

	}

	@Required

	public void setDbUtils(final DatabaseUtils dbUtils) {

		this.dbUtils = dbUtils;

	}

	public NotificationDispatcher getDispatcher() {

		return dispatcher;

	}

	@Required

	public void setDispatcher(final NotificationDispatcher dispatcher) {

		this.dispatcher = dispatcher;

	}

}