JGroupsConfigurator

package net.sf.tinyjee.jgroups.config;

import net.sf.tinyjee.Prioritizeable;
import net.sf.tinyjee.Server;
import net.sf.tinyjee.config.Configurable;
import net.sf.tinyjee.config.Connection;
import net.sf.tinyjee.config.Property;
import net.sf.tinyjee.jgroups.JGroupsLoggingReceiverAdapter;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.jgroups.Channel;
import org.jgroups.JChannel;
import org.jgroups.auth.MD5Token;
import org.jgroups.protocols.AUTH;
import org.jgroups.protocols.ENCRYPT;
import org.jgroups.protocols.TP;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.ProtocolStack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

import static net.sf.tinyjee.jgroups.config.JGroupsConnection.*;
import static net.sf.tinyjee.jgroups.config.JGroupsConnection.DETAULT_ENCRYPTION_ALGORITHM;
import static net.sf.tinyjee.jgroups.config.JGroupsConnection.KEY_JGROUPS_CLUSTER_NAME;
import static org.jgroups.stack.ProtocolStack.ABOVE;
import static org.jgroups.stack.ProtocolStack.BELOW;

/**
 * Hooks into the Server LifeCycle and Configuration
 *
 * @author Juergen_Kellerer, 2011-02-10
 * @version 1.0
 */
public class JGroupsConfigurator extends AbstractLifeCycle implements Configurable, Prioritizeable {

	private static final Logger log = LoggerFactory.getLogger(JGroupsConfigurator.class);

	private static final String JGROUPS_UDP_MCAST_ADDR = "jgroups.udp.mcast_addr";
	private static final String JGROUPS_UDP_IP_TTL = "jgroups.udp.ip_ttl";
	private static final String DEFAULT_JGROUPS_MULTICAST_IP_TTL = System.getProperty(JGROUPS_UDP_IP_TTL, "");
	private static final String DEFAULT_JGROUPS_MULTICAST_ADDRESS = System.getProperty(JGROUPS_UDP_MCAST_ADDR, "");

	private static String machineId = System.getProperty("tinyjee.jgroups.machine.id");

	static String getMachineId() {
		if (machineId == null) {
			try {
				machineId = InetAddress.getLocalHost().getCanonicalHostName();
			} catch (UnknownHostException e) {
				log.warn("Cannot create properly named JGroups channels as 'tinyjee.jgroups.machine.id' was not " +
						"specified as system property and auto-detection failed with.", e);
				machineId = "";
			}
		}
		return machineId;
	}

	private final static Object lock = new Object();
	private final Queue<JChannel> pendingChannels = new ConcurrentLinkedQueue<JChannel>();
	private final Map<JChannel, Connection> channels = new IdentityHashMap<JChannel, Connection>();

	/**
	 * {@inheritDoc}
	 */
	public int getPriority() {
		// Make sure this configurable is scheduled before others.
		return 1;
	}

	/**
	 * {@inheritDoc}
	 */
	public void configure() {
		for (Connection connection : JGroupsConnection.getJGroupsConnections())
			synchronized (lock) {
				try {
					// Export some parameters to the sys environment to allow
					// their usage inside the jgroups properties.
					exportConnectionAddressAndPort(connection);

					// Create a new channel
					String properties = connection.getPropertyValue(KEY_JGROUPS_PROPERTIES, DEFAULT_PROPERTIES);
					if (!properties.endsWith(".xml") && !properties.contains("://"))
						properties += ".xml";
					JChannel channel = new JChannel(properties);
					channel.setName(connection.getName() + "@" + getMachineId());

					// Configure a view logger (used until the receiver is changed to something else)
					channel.setReceiver(new JGroupsLoggingReceiverAdapter());

					// Setup the logging of discarded messages.
					for (Protocol protocol : channel.getProtocolStack().getProtocols()) {
						if (protocol instanceof TP) {
							Property property = connection.getProperty(KEY_JGROUPS_LOG_DISCARDED_MESSAGES);
							if (property != null)
								((TP) protocol).setLogDiscardMessages(property.getValue().getBoolean());
							if (connection.getPortNumber() != -1)
								((TP) protocol).setBindPort(connection.getPortNumber());
						}
					}

					try {
						addAuthProtocolIfNeeded(connection, channel);
					} catch (Exception e) {
						log.error("TJEE-00840:Failed to enable pre-shared key authentication on " +
								"jgroups connection " + toString(connection) + ". " +
								"Not initializing the channel to prevent data integrity problems.", e);
						continue;
					}

					try {
						addEncryptionProtocolIfNeeded(connection, channel);
					} catch (Exception e) {
						log.error("TJEE-00890:Failed to enable encryption on " +
								"jgroups connection " + toString(connection) + ". " +
								"Not initializing the channel to prevent data integrity problems.", e);
						continue;
					}

					channels.put(channel, connection);
					pendingChannels.offer(channel);

					connection.bindRelatedInstanceToJNDI(channel);

				} catch (Exception e) {
					log.error("TJEE-00790:Failed creating the jgroups channel for connection " +
							toString(connection), e);
				}
			}

		// Make sure we connect and disconnect the jgroups channels with the Server's life-cycle.
		Server.addToLifeCycle(this);
	}

	/**
	 * {@inheritDoc}
	 */
	public void reconfigure() throws Exception {
		doStop();
		configure();
		doStart();
	}

	private void exportConnectionAddressAndPort(Connection connection) {
		updateSystemProperty("jgroups.bind_address", connection.getAddressValue());
		updateSystemProperty(JGROUPS_UDP_IP_TTL,
				connection.getPropertyValue(KEY_JGROUPS_MULTICAST_IP_TTL, DEFAULT_JGROUPS_MULTICAST_IP_TTL));
		updateSystemProperty(JGROUPS_UDP_MCAST_ADDR,
				connection.getPropertyValue(KEY_JGROUPS_MULTICAST_ADDRESS, DEFAULT_JGROUPS_MULTICAST_ADDRESS));
		if (connection.getPortNumber() != -1)
			System.setProperty("jgroups.port", String.valueOf(connection.getPortNumber()));
	}

	private void updateSystemProperty(String key, String value) {
		if (value != null && !value.isEmpty())
			System.setProperty(key, value);
		else
			System.getProperties().remove(key);
	}

	private void addAuthProtocolIfNeeded(Connection connection, JChannel channel) throws Exception {
		final String authToken = connection.getPropertyValue(KEY_JGROUPS_AUTHENTICATION_KEY, "");
		if (!authToken.isEmpty()) {
			final ProtocolStack stack = channel.getProtocolStack();

			final boolean isAuthConfigured = null != stack.findProtocol(AUTH.class);
			if (!isAuthConfigured) {
				AUTH auth = new AUTH();
				MD5Token token = new MD5Token(authToken);
				token.setAuth(auth);
				auth.setValue("auth_token", token);

				// Placing auth below GMS => joining a cluster is only possible when AUTH succeeded.
				Protocol anchor = findAnchorProtocol(connection, stack, "GMS");
				stack.insertProtocol(auth, BELOW, anchor.getClass());
				stack.init();
			}

			if (log.isDebugEnabled())
				if (isAuthConfigured) {
					log.debug("Not changing existing auth configuration " +
							"used for jgroups connection '{}'", toString(connection));
				} else {
					log.debug("Configuring SHA1 pre-shared key authentication " +
							"for jgroups connection '{}'", toString(connection));
				}
		}
	}

	private void addEncryptionProtocolIfNeeded(Connection connection, JChannel channel) throws Exception {
		final String encryptionAlgorithm = connection.getPropertyValue(
				KEY_JGROUPS_ENCRYPTION_ALGORITHM, DETAULT_ENCRYPTION_ALGORITHM);
		if (!DETAULT_ENCRYPTION_ALGORITHM.equalsIgnoreCase(encryptionAlgorithm)) {
			final ProtocolStack stack = channel.getProtocolStack();

			final boolean isEncryptionConfigured = null != stack.findProtocol(ENCRYPT.class);
			if (!isEncryptionConfigured) {
				ENCRYPT encrypt = new ENCRYPT();
				encrypt.setValue("symAlgorithm", encryptionAlgorithm);

				// Placing encryption above auth  => keys are exchanged only when AUTH succeeded.
				Protocol anchor = findAnchorProtocol(connection, stack, "AUTH");
				stack.insertProtocol(encrypt, ABOVE, anchor.getClass());
				stack.init();
			}

			if (log.isDebugEnabled())
				if (isEncryptionConfigured) {
					log.debug("Not changing existing encryption configuration " +
							"used for jgroups connection '{}'", toString(connection));
				} else {
					log.debug("Configuring '{}' encryption " +
							"for jgroups connection '{}'", encryptionAlgorithm, toString(connection));
				}
		}
	}

	private Protocol findAnchorProtocol(
			Connection connection, ProtocolStack stack, String protocolName) throws Exception {
		Protocol anchor = null;
		for (Protocol protocol : stack.getProtocols())
			if (protocol.getName().contains(protocolName))
				anchor = protocol;

		if (anchor == null) {
			throw new Exception("Cannot configure jgroups connection " + toString(connection) +
					" with the configured features (auth or encryption) as no " + protocolName +
					" protocol was specified inside the given jgroups protocol stack.");
		}
		return anchor;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	protected void doStart() throws Exception {
		if (connectionThread != null)
			connectionThread.interrupt();

		connectPendingChannels();

		if (!pendingChannels.isEmpty()) {
			connectionThread = new Thread(connectPendingChannelsClosure, "Asynchronous JGroups Connection Thread");
			connectionThread.setDaemon(true);
			connectionThread.start();
		}
	}

	private Thread connectionThread;
	private Runnable connectPendingChannelsClosure = new Runnable() {
		public void run() {
			int retryTime = 0;
			while (!pendingChannels.isEmpty()) {
				try {
					Thread.sleep(TimeUnit.MINUTES.toMillis(retryTime++));
				} catch (InterruptedException e) {
					return;
				}

				connectPendingChannels();
			}
		}
	};

	private synchronized void connectPendingChannels() {
		final List<JChannel> failed = new ArrayList<JChannel>();

		JChannel channel;
		while ((channel = pendingChannels.poll()) != null) {
			final Connection connection = channels.get(channel);

			String clusterName = connection.getPropertyValue(KEY_JGROUPS_CLUSTER_NAME, "");
			if (clusterName.isEmpty()) {
				log.error("TJEE-00810:Cannot start jgroups channel. " +
						"A cluster name is missing for the connection '{}'.", toString(connection));
			}

			try {
				if (log.isDebugEnabled())
					log.debug("About to connect jgroups channel '{}'", toString(channel));

				channel.connect(clusterName);
				log.info("TJEE-00860:Successfully connected the jgroups channel '{}'", toString(channel));

				if (channel.getReceiver() instanceof JGroupsLoggingReceiverAdapter)
					((JGroupsLoggingReceiverAdapter) channel.getReceiver()).setLocalAddress(channel.getAddress());
			} catch (Exception e) {
				log.error("TJEE-00820:Failed connecting jgroups channel '" + channel.getName() +
						"' with cluster '" + clusterName + "'", e);
				failed.add(channel);
			}
		}

		pendingChannels.addAll(failed);
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	protected void doStop() throws Exception {
		pendingChannels.clear();
		for (Map.Entry<JChannel, Connection> entry : channels.entrySet()) {
			final JChannel channel = entry.getKey();
			final Connection connection = entry.getValue();

			try {
				channel.disconnect();
				channel.close();
				log.info("TJEE-00870:Successfully closed the jgroups channel '{}'", toString(channel));
				connection.unbindRelatedInstance();
			} catch (Exception e) {
				log.error("TJEE-00800:Failed closing the jgroups channel '" + toString(channel) + "'.", e);
			}
		}

		channels.clear();
	}

	private static String toString(Connection connection) {
		return connection.getName() + "/" +
				connection.getPropertyValue(KEY_JGROUPS_CLUSTER_NAME, "!!NO-CLUSTER-NAME!!");
	}

	private static String toString(Channel channel) {
		return channel.getName() + "/" + (channel.isConnected() ?
				channel.getClusterName() + "/" + channel.getAddress() :
				"unconnected");
	}
}