JvmCluster.java

package org.microspace.transport.specific;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

import org.microspace.failover.FailoverState;
import org.microspace.util.MicroLogger;

/**
 * Cluster for Jvm Based Transports.
 * 
 * @author Gaspar Sinai - {@literal gaspar.sinai@microspace.org}
 * @version 2016-06-26
 */

public class JvmCluster {
	
	final MicroLogger log = new MicroLogger (JvmCluster.class);
	final String name;
	
	final Map<Integer, JvmTransport> transports = new HashMap<Integer, JvmTransport> ();
	
	public JvmCluster(String name) {
		this.name = name;
	}

	public String getName() {
		return name;
	}
	
	void distribute (int from, Serializable msg, Integer to) {
		if (to != null) {
			if (from == to) {
				throw new IllegalArgumentException ("Distributed to ourself");
			}
			JvmTransport t = transports.get(to);
			t.distribute (from, msg);
			return;
		}
		for (JvmTransport t : transports.values()) {
			if (t.getId() == from) continue;
			t.distribute (from, msg);
		}
	}
	
	synchronized public void add (JvmTransport transport) {
		if (transports.containsKey(transport.getId())) {
			throw new IllegalArgumentException ("JvmTransport already exists");
		}
		Integer oldPrimary = getOldest ();
		transports.put(transport.getId(), transport);
		Integer newPrimary = getOldest ();
		failover (oldPrimary, newPrimary);
		if (transport.getFailoverState() == FailoverState.INITIALIZING) {
			transport.failoverStateChanged(FailoverState.BACKUP);
		}
	}
	
	synchronized public void remove (JvmTransport transport) {
		if (!transports.containsKey(transport.getId())) {
			return;
		}
		Integer oldPrimary = getOldest ();
		transports.remove(transport.getId());
		Integer newPrimary = getOldest ();
		failover (oldPrimary, newPrimary);
		if (transport.getFailoverState() != FailoverState.BACKUP) {
			transport.failoverStateChanged(FailoverState.BACKUP);
			
		}
	}
	
	private void failover (Integer from, Integer to) {
		log.info("Failover from $* to $*", from, to);
		if (from != null) {
			JvmTransport ft = transports.get(from);
			if (!from.equals(to) && ft != null) {
				ft.failoverStateChanged(FailoverState.BACKUP);
			}
		}
		if (to != null) {
			JvmTransport ft = transports.get(to);
			if (!to.equals(from)) {
				ft.failoverStateChanged(FailoverState.PRIMARY);
			}
		}
	}
	
	
	private Integer getOldest() {
		Integer oldest = null;
		for (int id : transports.keySet()) {
			if (oldest == null) {
				oldest = id;
			} else if (id < oldest) {
				oldest = id;
			}
		}
		return oldest;
	}
	
	public JvmTransport getOldestTransport() {
		Integer old = getOldest();
		if (old == null) return null;
		return transports.get(old);
	}
	
}