diff --git a/src/main/java/com/avaje/ebean/EbeanServerFactory.java b/src/main/java/com/avaje/ebean/EbeanServerFactory.java
index 290def9fb..d6326b5ea 100644
--- a/src/main/java/com/avaje/ebean/EbeanServerFactory.java
+++ b/src/main/java/com/avaje/ebean/EbeanServerFactory.java
@@ -133,7 +133,7 @@ public class EbeanServerFactory {
/**
* Create the container instance using the configuration.
*/
- private static SpiContainer createContainer(ContainerConfig containerConfig) {
+ protected static SpiContainer createContainer(ContainerConfig containerConfig) {
String implClassName = System.getProperty("ebean.container", DEFAULT_CONTAINER);
diff --git a/src/main/java/com/avaje/ebean/config/ContainerConfig.java b/src/main/java/com/avaje/ebean/config/ContainerConfig.java
index f25a97759..c8f243645 100644
--- a/src/main/java/com/avaje/ebean/config/ContainerConfig.java
+++ b/src/main/java/com/avaje/ebean/config/ContainerConfig.java
@@ -1,7 +1,5 @@
package com.avaje.ebean.config;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Properties;
/**
@@ -12,453 +10,51 @@ import java.util.Properties;
*/
public class ContainerConfig {
+ protected boolean clusterActive;
+
+ protected Properties properties;
+
/**
- * Communication mode used for clustering.
+ * Return true if clustering is active.
*/
- public enum ClusterMode {
-
- /**
- * No clustering.
- */
- NONE,
-
- /**
- * Use Multicast networking for cluster wide communication.
- */
- MULTICAST,
-
- /**
- * Use TCP Sockets for cluster wide communication.
- */
- SOCKET
+ public boolean isClusterActive() {
+ return clusterActive;
}
/**
- * The cluster mode to use.
+ * Set to true for clustering to be active.
*/
- ClusterMode mode = ClusterMode.NONE;
-
- /**
- * Configuration if using TCP sockets for clustering communication.
- */
- SocketConfig socketConfig = new SocketConfig();
-
- /**
- * Configuration if using Multicast for clustering communication.
- */
- MulticastConfig multicastConfig = new MulticastConfig();
-
-
- // -------------------------------------------------------------------------------------------
- // MulticastConfig
-
- /**
- * The configuration for clustering using Multicast networking.
- */
- public static class MulticastConfig {
-
- int managerSleepMillis = 80;
- int lastSendTimeFreqSecs = 300;//5mins
- int lastStatusTimeFreqSecs = 600;//10mins
- int maxResendOutgoingAttempts = 200;
- int maxResendIncomingRequests = 50;
-
- int listenPort;
- String listenAddress;
- int sendPort;
- String sendAddress;
-
- // Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507
- int maxSendPacketSize = 1500;
-
- // Whether to send packets even when there are no other members online
- boolean sendWithNoMembers = true;
-
- // When multiple instances are on same box you need to broadcast back locally
- boolean disableLoopback;
- int listenTimeToLive = -1;
- int listenTimeout = 1000;
- int listenBufferSize = 65500;
- // For multihomed environment the address the listener should bind to
- String listenBindAddress;
-
- /**
- * Return the manager sleep millis.
- */
- public int getManagerSleepMillis() {
- return managerSleepMillis;
- }
-
- /**
- * Set the manager sleep millis.
- */
- public void setManagerSleepMillis(int managerSleepMillis) {
- this.managerSleepMillis = managerSleepMillis;
- }
-
- /**
- * Return the last send time frequency.
- */
- public int getLastSendTimeFreqSecs() {
- return lastSendTimeFreqSecs;
- }
-
- /**
- * Set the last send time frequency.
- */
- public void setLastSendTimeFreqSecs(int lastSendTimeFreqSecs) {
- this.lastSendTimeFreqSecs = lastSendTimeFreqSecs;
- }
-
- /**
- * Return the last status time frequency.
- */
- public int getLastStatusTimeFreqSecs() {
- return lastStatusTimeFreqSecs;
- }
-
- /**
- * Set the last status time frequency.
- */
- public void setLastStatusTimeFreqSecs(int lastStatusTimeFreqSecs) {
- this.lastStatusTimeFreqSecs = lastStatusTimeFreqSecs;
- }
-
- /**
- * Return the maximum number of times we will try to re-send a given packet before giving up sending
- */
- public int getMaxResendOutgoingAttempts() {
- return maxResendOutgoingAttempts;
- }
-
- /**
- * Set the maximum retry attempts for outgoing messages.
- */
- public void setMaxResendOutgoingAttempts(int maxResendOutgoingAttempts) {
- this.maxResendOutgoingAttempts = maxResendOutgoingAttempts;
- }
-
- /**
- * Return the maximum number of times we will ask for a packet to be resent to us before giving up asking.
- */
- public int getMaxResendIncomingRequests() {
- return maxResendIncomingRequests;
- }
-
- /**
- * Set the maximum retry attempts for incoming messages.
- */
- public void setMaxResendIncomingRequests(int maxResendIncomingRequests) {
- this.maxResendIncomingRequests = maxResendIncomingRequests;
- }
-
- /**
- * Return the listen port.
- */
- public int getListenPort() {
- return listenPort;
- }
-
- /**
- * Set the listen port.
- */
- public void setListenPort(int port) {
- this.listenPort = port;
- }
-
- /**
- * Return the listen address.
- */
- public String getListenAddress() {
- return listenAddress;
- }
-
- /**
- * Set the listen address.
- */
- public void setListenAddress(String listenAddress) {
- this.listenAddress = listenAddress;
- }
-
- /**
- * Return the send port.
- */
- public int getSendPort() {
- return sendPort;
- }
-
- /**
- * Set the send port.
- */
- public void setSendPort(int sendPort) {
- this.sendPort = sendPort;
- }
-
- /**
- * Return the send address.
- */
- public String getSendAddress() {
- return sendAddress;
- }
-
- /**
- * Set the send address.
- */
- public void setSendAddress(String sendAddress) {
- this.sendAddress = sendAddress;
- }
-
- /**
- * Return the maximum send packet size.
- */
- public int getMaxSendPacketSize() {
- return maxSendPacketSize;
- }
-
- /**
- * Set the maximum send packet size. Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507.
- */
- public void setMaxSendPacketSize(int maxSendPacketSize) {
- this.maxSendPacketSize = maxSendPacketSize;
- }
-
- /**
- * Return true if send messages when no other members in the cluster are up.
- */
- public boolean isSendWithNoMembers() {
- return sendWithNoMembers;
- }
-
- /**
- * Set true if send messages when no other members in the cluster are up.
- */
- public void setSendWithNoMembers(boolean sendWithNoMembers) {
- this.sendWithNoMembers = sendWithNoMembers;
- }
-
- /**
- * Return true if loopback is disabled. When multiple instances are on same box you need to broadcast back locally.
- */
- public boolean isDisableLoopback() {
- return disableLoopback;
- }
-
- /**
- * Set if loopback is disabled. When multiple instances are on same box you need to broadcast back locally.
- */
- public void setDisableLoopback(boolean disableLoopback) {
- this.disableLoopback = disableLoopback;
- }
-
- /**
- * Return the listen time to live.
- */
- public int getListenTimeToLive() {
- return listenTimeToLive;
- }
-
- /**
- * Set the listen time to live.
- */
- public void setListenTimeToLive(int listenTimeToLive) {
- this.listenTimeToLive = listenTimeToLive;
- }
-
- /**
- * Return the listen timeout.
- */
- public int getListenTimeout() {
- return listenTimeout;
- }
-
- /**
- * set the listen timeout.
- */
- public void setListenTimeout(int listenTimeout) {
- this.listenTimeout = listenTimeout;
- }
-
- /**
- * Return the listen buffer size.
- */
- public int getListenBufferSize() {
- return listenBufferSize;
- }
-
- /**
- * Set the listen buffer size.
- */
- public void setListenBufferSize(int listenBufferSize) {
- this.listenBufferSize = listenBufferSize;
- }
-
- /**
- * Return the listener bind address (optional). For multihomed environment the address the listener should bind to.
- */
- public String getListenBindAddress() {
- return listenBindAddress;
- }
-
- /**
- * Set the listener bind address (optional). For multihomed environment the address the listener should bind to.
- */
- public void setListenBindAddress(String listenBindAddress) {
- this.listenBindAddress = listenBindAddress;
- }
+ public void setClusterActive(boolean clusterActive) {
+ this.clusterActive = clusterActive;
}
- // -------------------------------------------------------------------------------------------
- // SocketConfig
-
/**
- * Configuration for clustering using TCP sockets.
- *
- * This is good for when there are relatively small number of cluster members.
+ * Return the deployment properties.
*/
- public static class SocketConfig {
-
- /**
- * This local server in host:port format.
- */
- String localHostPort;
-
- /**
- * All the cluster members in host:port format.
- */
- List members = new ArrayList();
-
- /**
- * core threads for the associated thread pool.
- */
- int coreThreads = 2;
-
- /**
- * Max threads for the associated thread pool.
- */
- int maxThreads = 16;
-
- String threadPoolName = "EbeanCluster";
-
- /**
- * Return the host and port for this server instance.
- */
- public String getLocalHostPort() {
- return localHostPort;
- }
-
- /**
- * Set the host and port for this server instance.
- */
- public void setLocalHostPort(String localHostPort) {
- this.localHostPort = localHostPort;
- }
-
- /**
- * Return all the host and port for all the members of the cluster.
- */
- public List getMembers() {
- return members;
- }
-
- /**
- * Set all the host and port for all the members of the cluster.
- */
- public void setMembers(List members) {
- this.members = members;
- }
-
- /**
- * Return the number of core threads to use.
- */
- public int getCoreThreads() {
- return coreThreads;
- }
-
- /**
- * Set the number of core threads to use.
- */
- public void setCoreThreads(int coreThreads) {
- this.coreThreads = coreThreads;
- }
-
- /**
- * Return the number of max threads to use.
- */
- public int getMaxThreads() {
- return maxThreads;
- }
-
- /**
- * Set the number of max threads to use.
- */
- public void setMaxThreads(int maxThreads) {
- this.maxThreads = maxThreads;
- }
-
- /**
- * Return the thread pool name.
- */
- public String getThreadPoolName() {
- return threadPoolName;
- }
-
- /**
- * Set the thread pool name.
- */
- public void setThreadPoolName(String threadPoolName) {
- this.threadPoolName = threadPoolName;
- }
+ public Properties getProperties() {
+ return properties;
}
- // -------------------------------------------------------------------------------------------
- // Members
+ /**
+ * Set the deployment properties.
+ */
+ public void setProperties(Properties properties) {
+ this.properties = properties;
+ }
/**
* Load the settings from properties.
*/
public void loadFromProperties(Properties properties) {
- //TODO
+ this.properties = properties;
+ this.clusterActive = getProperty(properties, "ebean.cluster.active", clusterActive);
}
/**
- * Return the cluster mode.
+ * Return the boolean property setting.
*/
- public ClusterMode getMode() {
- return mode;
+ protected boolean getProperty(Properties properties, String key, boolean defaultValue) {
+ return "true".equalsIgnoreCase(properties.getProperty(key, Boolean.toString(defaultValue)));
}
- /**
- * Set the cluster mode.
- */
- public void setMode(ClusterMode mode) {
- this.mode = mode;
- }
-
- /**
- * Return the socket communication configuration.
- */
- public SocketConfig getSocketConfig() {
- return socketConfig;
- }
-
- /**
- * Set the socket communication configuration.
- */
- public void setSocketConfig(SocketConfig socketConfig) {
- this.socketConfig = socketConfig;
- }
-
- /**
- * Return the multicast communication configuration.
- */
- public MulticastConfig getMulticastConfig() {
- return multicastConfig;
- }
-
- /**
- * Set the multicast communication configuration.
- */
- public void setMulticastConfig(MulticastConfig multicastConfig) {
- this.multicastConfig = multicastConfig;
- }
}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java
index 414c0bfaf..8bf919048 100644
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java
+++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java
@@ -12,7 +12,7 @@ public interface ClusterBroadcast {
* Inform the other cluster members that this instance has come online and
* start any listeners etc.
*/
- void startup(ClusterManager clusterManager);
+ void startup();
/**
* Inform the other cluster members that this instance is leaving and
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java
new file mode 100644
index 000000000..af303d919
--- /dev/null
+++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java
@@ -0,0 +1,14 @@
+package com.avaje.ebeaninternal.server.cluster;
+
+import java.util.Properties;
+
+/**
+ * Factory to create the cluster broadcast service.
+ */
+public interface ClusterBroadcastFactory {
+
+ /**
+ * Create the cluster transport with the manager and deployment properties.
+ */
+ ClusterBroadcast create(ClusterManager manager, Properties properties);
+}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java
index 67318d2d3..30af29e53 100644
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java
+++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java
@@ -2,12 +2,12 @@ package com.avaje.ebeaninternal.server.cluster;
import com.avaje.ebean.EbeanServer;
import com.avaje.ebean.config.ContainerConfig;
-import com.avaje.ebeaninternal.server.cluster.mcast.McastClusterManager;
-import com.avaje.ebeaninternal.server.cluster.socket.SocketClusterBroadcast;
import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Iterator;
+import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -15,6 +15,8 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class ClusterManager {
+ private static final Logger clusterLogger = LoggerFactory.getLogger("org.avaje.ebean.Cluster");
+
private static final Logger logger = LoggerFactory.getLogger(ClusterManager.class);
private final ConcurrentHashMap serverMap = new ConcurrentHashMap();
@@ -25,30 +27,33 @@ public class ClusterManager {
private boolean started;
- public ClusterManager(ContainerConfig containerConfig) {
-
- ContainerConfig.ClusterMode mode = containerConfig.getMode();
- try {
- switch (mode) {
- case SOCKET: {
- this.broadcast = new SocketClusterBroadcast(containerConfig);
- break;
- }
- case MULTICAST: {
- this.broadcast = new McastClusterManager(containerConfig);
- break;
- }
- default: {
- this.broadcast = null;
- }
- }
-
- } catch (Exception e) {
- logger.error("Error initialising ClusterManager type [" + mode + "]", e);
- throw new RuntimeException(e);
+ public ClusterManager(ContainerConfig config) {
+ if (!config.isClusterActive()) {
+ broadcast = null;
+ } else {
+ ClusterBroadcastFactory factory = createFactory();
+ broadcast = factory.create(this, config.getProperties());
}
}
+ /**
+ * Return the ClusterTransportFactory via ServiceLoader.
+ */
+ private ClusterBroadcastFactory createFactory() {
+
+ ServiceLoader load = ServiceLoader.load(ClusterBroadcastFactory.class);
+ ClusterBroadcastFactory factory = null;
+ Iterator iterator = load.iterator();
+ if (iterator.hasNext()) {
+ factory = iterator.next();
+ }
+ if (factory == null) {
+ throw new IllegalStateException("No ClusterTransportFactory found in classpath. "
+ + " Probably need to add the avaje-ebeanorm-cluster dependency");
+ }
+ return factory;
+ }
+
public void registerServer(EbeanServer server) {
synchronized (monitor) {
serverMap.put(server.getName(), server);
@@ -67,7 +72,7 @@ public class ClusterManager {
private void startup() {
started = true;
if (broadcast != null) {
- broadcast.startup(this);
+ broadcast.startup();
}
}
@@ -81,9 +86,12 @@ public class ClusterManager {
/**
* Send the message headers and payload to every server in the cluster.
*/
- public void broadcast(RemoteTransactionEvent remoteTransEvent) {
+ public void broadcast(RemoteTransactionEvent event) {
if (broadcast != null) {
- broadcast.broadcast(remoteTransEvent);
+ if (clusterLogger.isDebugEnabled()) {
+ clusterLogger.debug("sending: {}", event);
+ }
+ broadcast.broadcast(event);
}
}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java
deleted file mode 100644
index 46f4e148c..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import java.io.Serializable;
-
-/**
- * Simple holder of binary data.
- * Used to use Packet based serialisation of RemoteTransactionEvent
- * with simple Java Serialisation of the DataHolder.
- */
-public class DataHolder implements Serializable {
-
- private static final long serialVersionUID = 9090748723571322192L;
-
- private final byte[] data;
-
- public DataHolder(byte[] data) {
- this.data = data;
- }
-
- public byte[] getData() {
- return data;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java
deleted file mode 100644
index 1966a34f6..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java
+++ /dev/null
@@ -1,193 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import java.io.ByteArrayOutputStream;
-import java.io.DataInput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-/**
- * Represents the contents sent as a single DatagramPacket.
- *
- * The contents is typically multiple messages (ACK,PING etc) or all or part of
- * a RemoteTransactionEvent.
- *
- *
- * Due to the hard limit on the size of UDP packets a RemoteTransactionEvent
- * with lots of information could be broken up into multiple packets.
- *
- *
- * @author rbygrave
- */
-public class Packet {
-
- /**
- * A Packet that holds protocol messages like ACK, PING etc.
- */
- public static final short TYPE_MESSAGES = 1;
-
- /**
- * A Packet that holds TransactionEvent information such as Bean
- * and or Table IUD information.
- */
- public static final short TYPE_TRANSEVENT = 2;
-
- /**
- * The type of Packet.
- */
- protected final short packetType;
-
- /**
- * The PacketId.
- */
- protected final long packetId;
-
- /**
- * The timestamp the Packet was created.
- */
- protected final long timestamp;
-
- /**
- * The EbeanServer name this relates to if relevant.
- */
- protected final String serverName;
-
- protected ByteArrayOutputStream buffer;
- protected DataOutputStream dataOut;
- protected byte[] bytes;
-
- /**
- * The number of messages in this Packet.
- */
- private int messageCount;
-
- /**
- * The number of times this Packet was resent.
- */
- private int resendCount;
-
- /**
- * Create a Packet for writing messages to.
- */
- public static Packet forWrite(short packetType, long packetId, long timestamp, String serverName) throws IOException {
- return new Packet(true, packetType, packetId, timestamp, serverName);
- }
-
- /**
- * Create a Packet just reading the Header information.
- */
- public static Packet readHeader(DataInput dataInput) throws IOException {
-
- short packetType = dataInput.readShort();
- long packetId = dataInput.readLong();
- long timestamp = dataInput.readLong();
- String serverName = dataInput.readUTF();
-
- return new Packet(false, packetType, packetId, timestamp, serverName);
- }
-
- protected Packet(boolean write, short packetType, long packetId, long timestamp, String serverName) throws IOException {
- this.packetType = packetType;
- this.packetId = packetId;
- this.timestamp = timestamp;
- this.serverName = serverName;
- if (write) {
- this.buffer = new ByteArrayOutputStream();
- this.dataOut = new DataOutputStream(buffer);
- writeHeader();
- } else {
- this.buffer = null;
- this.dataOut = null;
- }
- }
-
- private void writeHeader() throws IOException {
- dataOut.writeShort(packetType);
- dataOut.writeLong(packetId);
- dataOut.writeLong(timestamp);
- dataOut.writeUTF(serverName);
- }
-
- public int incrementResendCount() {
- return resendCount++;
- }
-
- public short getPacketType() {
- return packetType;
- }
-
- public long getPacketId() {
- return packetId;
- }
-
- public long getTimestamp() {
- return timestamp;
- }
-
- public String getServerName() {
- return serverName;
- }
-
- public void writeEof() throws IOException {
- dataOut.writeBoolean(false);
- }
-
- public void read(DataInput dataInput) throws IOException {
- boolean more = dataInput.readBoolean();
- while (more) {
- int msgType = dataInput.readInt();
- readMessage(dataInput, msgType);
- // see if there is more information
- more = dataInput.readBoolean();
- }
- }
-
- /**
- * Overridden by more specific Packet implementations to read the messages.
- */
- protected void readMessage(DataInput dataInput, int msgType) throws IOException {
-
- }
-
- /**
- * Write a binary message to this packet returning true if there was
- * enough room to do so. Return false if the message was too large for
- * the remaining space left - in this case another Packet should be
- * created to put that message into.
- */
- public boolean writeBinaryMessage(BinaryMessage msg, int maxPacketSize) throws IOException {
-
- byte[] bytes = msg.getByteArray();
-
- if (messageCount > 0 && (bytes.length + buffer.size() > maxPacketSize)) {
- // we are actually going to ignore the maxPacketSize iff we have one
- // large message.
-
- // false = no more messages
- dataOut.writeBoolean(false);
- return false;
- }
- ++messageCount;
- // true = another message follows
- dataOut.writeBoolean(true);
- dataOut.write(bytes);
- return true;
- }
-
- public int getSize() {
- return getBytes().length;
- }
-
- /**
- * Return the Packet as raw bytes.
- */
- public byte[] getBytes() {
- if (bytes == null) {
- bytes = buffer.toByteArray();
- buffer = null;
- dataOut = null;
- }
- return bytes;
- }
-
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java
deleted file mode 100644
index cc43039f7..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java
+++ /dev/null
@@ -1,69 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import com.avaje.ebeaninternal.server.cluster.mcast.Message;
-import com.avaje.ebeaninternal.server.cluster.mcast.MessageAck;
-import com.avaje.ebeaninternal.server.cluster.mcast.MessageControl;
-import com.avaje.ebeaninternal.server.cluster.mcast.MessageResend;
-
-import java.io.DataInput;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * A Packet that contains Ack, Resend and Control messages.
- *
- * @author rbygrave
- */
-public class PacketMessages extends Packet {
-
- private final ArrayList messages;
-
- public static PacketMessages forWrite(long packetId, long timestamp, String serverName) throws IOException {
- return new PacketMessages(true, packetId, timestamp, serverName);
- }
-
- public static PacketMessages forRead(Packet header) throws IOException {
- return new PacketMessages(header);
- }
-
- private PacketMessages(boolean write, long packetId, long timestamp, String serverName) throws IOException {
- super(write, TYPE_MESSAGES, packetId, timestamp, serverName);
- this.messages = null;
- }
-
- private PacketMessages(Packet header) throws IOException {
- super(false, TYPE_MESSAGES, header.packetId, header.timestamp, header.serverName);
- this.messages = new ArrayList();
- }
-
- /**
- * Return the messages contained in this Packet.
- */
- public List getMessages() {
- return messages;
- }
-
- /**
- * Read the messages (Ack, Resend or Control) contained in this packet.
- */
- protected void readMessage(DataInput dataInput, int msgType) throws IOException {
-
- switch (msgType) {
- case BinaryMessage.TYPE_MSGCONTROL:
- messages.add(MessageControl.readBinaryMessage(dataInput));
- break;
-
- case BinaryMessage.TYPE_MSGACK:
- messages.add(MessageAck.readBinaryMessage(dataInput));
- break;
-
- case BinaryMessage.TYPE_MSGRESEND:
- messages.add(MessageResend.readBinaryMessage(dataInput));
- break;
-
- default:
- throw new RuntimeException("Invalid Transaction msgType " + msgType);
- }
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java
deleted file mode 100644
index a4dac5d2b..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java
+++ /dev/null
@@ -1,64 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import com.avaje.ebeaninternal.api.SpiEbeanServer;
-import com.avaje.ebeaninternal.api.TransactionEventTable.TableIUD;
-import com.avaje.ebeaninternal.server.transaction.BeanPersistIds;
-import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
-
-import java.io.DataInput;
-import java.io.IOException;
-
-/**
- * A Packet holding TransactionEvent data.
- *
- * Due to the hard limit for UDP packet sizes a RemoteTransactionEvent
- * is actually broken up into smaller messages.
- *
- */
-public class PacketTransactionEvent extends Packet {
-
- private final SpiEbeanServer server;
-
- private final RemoteTransactionEvent event;
-
- public static PacketTransactionEvent forWrite(long packetId, long timestamp, String serverName) throws IOException {
- return new PacketTransactionEvent(true, packetId, timestamp, serverName);
- }
-
- private PacketTransactionEvent(boolean write, long packetId, long timestamp, String serverName) throws IOException {
- super(write, TYPE_TRANSEVENT, packetId, timestamp, serverName);
- this.server = null;
- this.event = null;
- }
-
- private PacketTransactionEvent(Packet header, SpiEbeanServer server) throws IOException {
- super(false, TYPE_TRANSEVENT, header.packetId, header.timestamp, header.serverName);
- this.server = server;
- this.event = new RemoteTransactionEvent(server);
- }
-
- public static PacketTransactionEvent forRead(Packet header, SpiEbeanServer server) throws IOException {
- return new PacketTransactionEvent(header, server);
- }
-
- public RemoteTransactionEvent getEvent() {
- return event;
- }
-
- protected void readMessage(DataInput dataInput, int msgType) throws IOException {
-
- switch (msgType) {
- case BinaryMessage.TYPE_BEANIUD:
- event.addBeanPersistIds(BeanPersistIds.readBinaryMessage(server, dataInput));
- break;
-
- case BinaryMessage.TYPE_TABLEIUD:
- event.addTableIUD(TableIUD.readBinaryMessage(dataInput));
- break;
-
- default:
- throw new RuntimeException("Invalid Transaction msgType " + msgType);
- }
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java
deleted file mode 100644
index bc3ee0775..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java
+++ /dev/null
@@ -1,160 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import com.avaje.ebeaninternal.server.cluster.mcast.Message;
-import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Creates Packets for either RemoteTransactionEvents or Messages (Ping, ACK,
- * Join, Leave etc).
- *
- * @author rbygrave
- */
-public class PacketWriter {
-
- private final PacketIdGenerator idGenerator;
- private final PacketBuilder messagesPacketBuilder;
- private final PacketBuilder transEventPacketBuilder;
-
- /**
- * Create a PacketWriter with an expected max packet size.
- *
- * In theory we would prefer to create packets up to the MTU size which for
- * Ethernet will likely be 1500. Note that the maxPacketSize is ignored for
- * large single messages.
- *
- */
- public PacketWriter(int maxPacketSize) {
- this.idGenerator = new PacketIdGenerator();
- this.messagesPacketBuilder = new PacketBuilder(maxPacketSize, idGenerator, new MessagesPacketFactory());
- this.transEventPacketBuilder = new PacketBuilder(maxPacketSize, idGenerator, new TransPacketFactory());
- }
-
- /**
- * Return the currentPacketId.
- */
- public long currentPacketId() {
- return idGenerator.currentPacketId();
- }
-
- /**
- * Create Packets for a given list of messages.
- *
- * Typically this creates a single Packet but there is a hard limit for UDP
- * packet sizes.
- *
- */
- public List write(boolean requiresAck, List messages) throws IOException {
-
- BinaryMessageList binaryMsgList = new BinaryMessageList();
- for (int i = 0; i < messages.size(); i++) {
- Message message = messages.get(i);
- message.writeBinaryMessage(binaryMsgList);
- }
- return messagesPacketBuilder.write(requiresAck, binaryMsgList, "");
- }
-
- /**
- * Create Packets for a given RemoteTransactionEvent.
- *
- * Typically this creates a single Packet but there is a hard limit for UDP
- * packet sizes.
- *
- */
- public List write(RemoteTransactionEvent transEvent) throws IOException {
-
- BinaryMessageList messageList = new BinaryMessageList();
-
- // split into reasonably small independent messages
- transEvent.writeBinaryMessage(messageList);
-
- return transEventPacketBuilder.write(true, messageList, transEvent.getServerName());
- }
-
- /**
- * Reuse the same packetIdCounter for building Packets for both Message and
- * RemoteTransactionEvent
- */
- private static class PacketIdGenerator {
-
- long packetIdCounter;
-
- public long nextPacketId() {
- return ++packetIdCounter;
- }
-
- public long currentPacketId() {
- return packetIdCounter;
- }
-
- }
-
- interface PacketFactory {
-
- Packet createPacket(long packetId, long timestamp, String serverName) throws IOException;
- }
-
- private static class TransPacketFactory implements PacketFactory {
-
- public Packet createPacket(long packetId, long timestamp, String serverName) throws IOException {
- return PacketTransactionEvent.forWrite(packetId, timestamp, serverName);
- }
- }
-
- private static class MessagesPacketFactory implements PacketFactory {
-
- public Packet createPacket(long packetId, long timestamp, String serverName) throws IOException {
- return PacketMessages.forWrite(packetId, timestamp, serverName);
- }
- }
-
- /**
- * Helper class for building Packets from messages or
- * RemoteTransactionEvents.
- */
- private static class PacketBuilder {
-
- private final PacketIdGenerator idGenerator;
- private final PacketFactory packetFactory;
- private final int maxPacketSize;
-
- private PacketBuilder(int maxPacketSize, PacketIdGenerator idGenerator, PacketFactory packetFactory) {
- this.maxPacketSize = maxPacketSize;
- this.idGenerator = idGenerator;
- this.packetFactory = packetFactory;
- }
-
- private List write(boolean requiresAck, BinaryMessageList messageList, String serverName)
- throws IOException {
-
- List list = messageList.getList();
-
- ArrayList packets = new ArrayList(1);
-
- long timestamp = System.currentTimeMillis();
-
- long packetId = requiresAck ? idGenerator.nextPacketId() : 0;
- Packet p = packetFactory.createPacket(packetId, timestamp, serverName);
-
- packets.add(p);
-
- for (int i = 0; i < list.size(); i++) {
- BinaryMessage binMsg = list.get(i);
- if (!p.writeBinaryMessage(binMsg, maxPacketSize)) {
- // didn't fit into the package so put into another packet
- packetId = requiresAck ? idGenerator.nextPacketId() : 0;
- p = packetFactory.createPacket(packetId, timestamp, serverName);
- packets.add(p);
- p.writeBinaryMessage(binMsg, maxPacketSize);
- }
- }
- p.writeEof();
-
- return packets;
-
- }
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java
deleted file mode 100644
index daab0f50d..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster;
-
-import com.avaje.ebeaninternal.api.SpiEbeanServer;
-import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.util.List;
-
-/**
- * Mechanism to convert RemoteTransactionEvent to/from byte[] content.
- */
-public abstract class SerialiseTransactionHelper {
-
- private static final Logger logger = LoggerFactory.getLogger(SerialiseTransactionHelper.class);
-
- private final PacketWriter packetWriter;
-
- public SerialiseTransactionHelper() {
- packetWriter = new PacketWriter(Integer.MAX_VALUE);
- }
-
- public abstract SpiEbeanServer getEbeanServer(String serverName);
-
- /**
- * Convert the RemoteTransactionEvent to byte[] content.
- */
- public DataHolder createDataHolder(RemoteTransactionEvent transEvent) throws IOException {
-
- List packetList = packetWriter.write(transEvent);
- if (packetList.size() != 1) {
- throw new RuntimeException("Always expecting 1 Packet but got " + packetList.size());
- }
- byte[] data = packetList.get(0).getBytes();
- return new DataHolder(data);
- }
-
- /**
- * Convert the byte[] content to RemoteTransactionEvent.
- */
- public RemoteTransactionEvent read(DataHolder dataHolder) throws IOException {
-
- ByteArrayInputStream bi = new ByteArrayInputStream(dataHolder.getData());
- DataInputStream dataInput = new DataInputStream(bi);
-
- Packet header = Packet.readHeader(dataInput);
-
- SpiEbeanServer server = getEbeanServer(header.getServerName());
- if (server == null) {
- logger.error("server [{}] not found/registered?", header.getServerName());
- }
-
- PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server);
- tranEventPacket.read(dataInput);
- return tranEventPacket.getEvent();
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java
deleted file mode 100644
index 447194c9c..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java
+++ /dev/null
@@ -1,41 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Holds a list of ACK and RESEND messages that should be sent out.
- */
-public class AckResendMessages {
-
- final ArrayList messages = new ArrayList();
-
- public String toString() {
- return messages.toString();
- }
-
- public int size() {
- return messages.size();
- }
-
- /**
- * Add a ACK message to send.
- */
- public void add(MessageAck ack) {
- messages.add(ack);
- }
-
- /**
- * Add a RESEND message to send.
- */
- public void add(MessageResend resend) {
- messages.add(resend);
- }
-
- /**
- * Return all the messages to be sent out.
- */
- public List getMessages() {
- return messages;
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java
deleted file mode 100644
index 854f3c712..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java
+++ /dev/null
@@ -1,51 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import java.util.HashMap;
-import java.util.List;
-
-/**
- * For this node this holds the ACK gotAllPoint for each member in the cluster.
- *
- * As we receive messages from other members of the cluster periodically we need
- * to send them ACK messages to say we got all the packets up to the gotAllPoint.
- *
- * Thread Safety note: Object only used by McastClusterBroadcast Manager thread.
- * So Single Threaded access.
- */
-public class IncomingPacketsLastAck {
-
- private final HashMap lastAckMap = new HashMap();
-
- public String toString() {
- return lastAckMap.values().toString();
- }
-
- /**
- * Remove a member of the cluster who has left.
- */
- public void remove(String memberHostPort) {
- lastAckMap.remove(memberHostPort);
- }
-
- /**
- * Get the last Ack point for a given member of the cluster.
- */
- public MessageAck getLastAck(String memberHostPort) {
- return lastAckMap.get(memberHostPort);
- }
-
- /**
- * For the ACK messages in AckResendMessages update the
- * last Ack packetId.
- */
- public void updateLastAck(AckResendMessages ackResendMessages) {
- List messages = ackResendMessages.getMessages();
- for (int i = 0; i < messages.size(); i++) {
- Message msg = messages.get(i);
- if (msg instanceof MessageAck) {
- MessageAck lastAck = (MessageAck) msg;
- lastAckMap.put(lastAck.getToHostPort(), lastAck);
- }
- }
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java
deleted file mode 100644
index 372bb88ef..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java
+++ /dev/null
@@ -1,271 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * For Incoming Packets remembers the packets we have received and processed.
- *
- * This determines the gotAllPoint per cluster member and identifies missing
- * packets (gap between gotAllPoint and gotMaxPoint).
- *
- *
- * This information is used by the managerThread so send ACK's for messages we
- * have received and RESEND messages to fill the missing packets we have
- * detected.
- *
- */
-public class IncomingPacketsProcessed {
-
- private final ConcurrentHashMap mapByMember = new ConcurrentHashMap();
-
- private final int maxResendIncoming;
-
- public IncomingPacketsProcessed(int maxResendIncoming) {
- this.maxResendIncoming = maxResendIncoming;
- }
-
- public void removeMember(String memberKey) {
- mapByMember.remove(memberKey);
- }
-
- /**
- * Return true if we should process this packet. Return false if we have
- * already processed the packet.
- */
- public boolean isProcessPacket(String memberKey, long packetId) {
-
- GotAllPoint memberPackets = getMemberPackets(memberKey);
- return memberPackets.processPacket(packetId);
- }
-
- /**
- * Build the list of ACK and RESEND messages that we should send out
- * to the other members of the cluster.
- */
- public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
-
- // Called by the McastClusterBroadcast manager thread
-
- AckResendMessages response = new AckResendMessages();
-
- for (GotAllPoint member : mapByMember.values()) {
-
- MessageAck lastAckMessage = lastAck.getLastAck(member.getMemberKey());
-
- member.addAckResendMessages(response, lastAckMessage);
- }
-
- return response;
- }
-
- private GotAllPoint getMemberPackets(String memberKey) {
-
- // This method is only called single threaded
- // by the listener thread so I'm happy that this
- // put into mapByMember is ok.
- GotAllPoint memberGotAllPoint = mapByMember.get(memberKey);
- if (memberGotAllPoint == null) {
- memberGotAllPoint = new GotAllPoint(memberKey, maxResendIncoming);
- mapByMember.put(memberKey, memberGotAllPoint);
- }
- return memberGotAllPoint;
- }
-
- /**
- * Keeps track of packets received from a particular member of the cluster.
- *
- * It notes the packetIds of the packets received and uses those to maintain
- * the 'gotAllPoint'. The 'gotAllPoint' is the packetId which we know we
- * received all the previous packets.
- *
- */
- public static class GotAllPoint {
-
- private static final Logger logger = LoggerFactory.getLogger(GotAllPoint.class);
-
- private final String memberKey;
- private final int maxResendIncoming;
-
- private long gotAllPoint;
-
- private long gotMaxPoint;
-
- /**
- * Packets received out of order.
- */
- private final ArrayList outOfOrderList = new ArrayList();
-
- private final HashMap resendCountMap = new HashMap();
-
- public GotAllPoint(String memberKey, int maxResendIncoming) {
- this.memberKey = memberKey;
- this.maxResendIncoming = maxResendIncoming;
- }
-
- /**
- * Add ACK and RESEND messages if required.
- */
- public void addAckResendMessages(AckResendMessages response, MessageAck lastAckMessage) {
-
- synchronized (this) {
- if (lastAckMessage != null && lastAckMessage.getGotAllPacketId() >= gotAllPoint) {
- // nothing has changed
- } else {
- // ACK that we have got every packet up to gotAllPoint
- response.add(new MessageAck(memberKey, gotAllPoint));
- }
-
- if (getMissingPacketCount() > 0) {
- // Ask for these Packets to be RESENT
- List missingPackets = getMissingPackets();
- response.add(new MessageResend(memberKey, missingPackets));
- }
- }
- }
-
- public String getMemberKey() {
- return memberKey;
- }
-
- public long getGotAllPoint() {
- synchronized (this) {
- return gotAllPoint;
- }
- }
-
- public long getGotMaxPoint() {
- synchronized (this) {
- return gotMaxPoint;
- }
- }
-
- private int getMissingPacketCount() {
- if (gotMaxPoint <= gotAllPoint) {
- if (!resendCountMap.isEmpty()) {
- resendCountMap.clear();
- }
- return 0;
- }
- return (int) (gotMaxPoint - gotAllPoint) - outOfOrderList.size();
- }
-
- public List getMissingPackets() {
-
- synchronized (this) {
- ArrayList missingList = new ArrayList();
-
- // this is not particularly efficient but expecting
- // the outOfOrderList to be relatively small
-
- boolean lostPacket = false;
-
- for (long i = gotAllPoint + 1; i < gotMaxPoint; i++) {
- Long packetId = i;
- if (!outOfOrderList.contains(packetId)) {
- if (incrementResendCount(packetId)) {
- // request this packet be resent
- missingList.add(packetId);
- } else {
- lostPacket = true;
- }
- }
- }
-
- if (lostPacket) {
- checkOutOfOrderList();
- }
-
- return missingList;
- }
- }
-
- /**
- * Return true if this packet has not yet exceeded the maxResendCount.
- */
- private boolean incrementResendCount(Long packetId) {
- Integer resendCount = resendCountMap.get(packetId);
- if (resendCount != null) {
- int i = resendCount + 1;
- if (i > maxResendIncoming) {
- // we are going to give up trying to get this packet now
- logger.warn("Exceeded maxResendIncoming[" + maxResendIncoming + "] for packet[" + packetId + "]. Giving up on requesting it.");
- resendCountMap.remove(packetId);
- outOfOrderList.add(packetId);
- return false;
- }
- resendCount = i;
- resendCountMap.put(packetId, resendCount);
- } else {
- resendCountMap.put(packetId, ONE);
- }
- return true;
- }
-
- private static final Integer ONE = 1;
-
- public boolean processPacket(long packetId) {
- synchronized (this) {
-
- if (gotAllPoint == 0) {
- gotAllPoint = packetId;
- return true;
- }
- if (packetId <= gotAllPoint) {
- // already processed this packet
- return false;
- }
-
- if (!resendCountMap.isEmpty()) {
- resendCountMap.remove(Long.valueOf(packetId));
- }
-
- if (packetId == gotAllPoint + 1) {
- gotAllPoint = packetId;
- } else {
- if (packetId > gotMaxPoint) {
- gotMaxPoint = packetId;
- }
- outOfOrderList.add(packetId);
- }
- checkOutOfOrderList();
- return true;
- }
- }
-
- private void checkOutOfOrderList() {
-
- if (outOfOrderList.size() == 0) {
- return;
- }
-
- boolean continueCheck;
- do {
- continueCheck = false;
- long nextPoint = gotAllPoint + 1;
-
- Iterator it = outOfOrderList.iterator();
- while (it.hasNext()) {
- Long id = it.next();
- if (id == nextPoint) {
- // we found the next one in the outOfOrderList
- it.remove();
- gotAllPoint = nextPoint;
- continueCheck = true;
- break;
- }
- }
- } while (continueCheck);
-
- }
-
- }
-
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java
deleted file mode 100644
index b949834a9..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java
+++ /dev/null
@@ -1,571 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebean.config.ContainerConfig;
-import com.avaje.ebeaninternal.api.SpiEbeanServer;
-import com.avaje.ebeaninternal.server.cluster.ClusterBroadcast;
-import com.avaje.ebeaninternal.server.cluster.ClusterManager;
-import com.avaje.ebeaninternal.server.cluster.Packet;
-import com.avaje.ebeaninternal.server.cluster.PacketWriter;
-import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.TreeSet;
-
-/**
- * Overall Manager of the Multicast Cluster communication for this instance.
- *
- * McastListener, McastSender and McastPacketControl are the main helpers to
- * this object.
- *
- *
- * This Manager (thread) periodically processes the ACK, Re-send and Control
- * messages. The McastListener is handling all the incoming packets and informs
- * this manager when interesting packets need to be processed by the Manager.
- *
- *
- * Other threads call {@link #broadcast(RemoteTransactionEvent)} to send
- * transaction even information.
- *
- */
-public class McastClusterManager implements ClusterBroadcast, Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(McastClusterManager.class);
-
- private ClusterManager clusterManager;
-
- private final Thread managerThread;
-
- /**
- * Helps co-ordinate packet information (Acks, Missing Packets etc).
- */
- private final McastPacketControl packageControl;
-
- /**
- * Listeners for incoming packets.
- */
- private final McastListener listener;
-
- /**
- * Sends packets out to the cluster.
- */
- private final McastSender localSender;
-
- /**
- * The localSenderHostPort is used to identify this instance in the cluster.
- */
- private final String localSenderHostPort;
-
- /**
- * Creates the Packets (byte[]) from Messages and RemoteTransactionEvent.
- */
- private final PacketWriter packetWriter;
-
- /**
- * List of Re-send messages that the managerThread needs to process.
- */
- private final ArrayList resendMessages = new ArrayList();
-
- /**
- * List of Control messages (Ping,PingResponse,Join,Leave) that the managerThread needs to process.
- */
- private final ArrayList controlMessages = new ArrayList();
-
- /**
- * Cache of outgoing messages that have not been ACK'ed by the other cluster members yet.
- */
- private final OutgoingPacketsCache outgoingPacketsCache = new OutgoingPacketsCache();
-
- /**
- * The last ACK we sent out to other members of the cluster.
- */
- private final IncomingPacketsLastAck incomingPacketsLastAck = new IncomingPacketsLastAck();
-
- /**
- * A limit of the number of times we will try to send out a given packet.
- * Once this is exceeded we will just drop that packet. Hopefully this does
- * not happen but we don't want to keep trying forever producing network
- * traffic.
- */
- private final int maxResendOutgoing;
-
- /**
- * Instead of ACK'ing immediately we periodically wake up and in a single
- * packet (typically) ACK all members of the cluster everything we got since
- * the last sleep time. More frequent ACK's means less memory consumption as
- * Packets are cleared from the outgoingPacketsCache quicker at the cost of
- * sending more packets.
- */
- private final long managerSleepMillis;
-
- /**
- * When true then packets are still sent out even when the cluster has no other online members.
- */
- private final boolean sendWithNoMembers;
-
- /**
- * The current minAcked packetId processed by the managerThread.
- * All packets before this have been ACK'ed by everyone in the cluster.
- */
- private long minAcked;
-
- /**
- * The min packetId that has been ACKed by all the members of the cluster according
- * to the McastListener. This will increase as the Listener receives ACK's and means
- * we can trim out Packets from the sent cache.
- */
- private long minAckedFromListener;
-
- /**
- * Start the groupSize at -1 so we have to wait until the Listener times out or gets
- * a control messages (Ping, PingResponse, Join, Leave etc) before we know how many
- * members of the group the listener knows about.
- *
- * Generally speaking we only care if the groupSize == 0 meaning there are no other
- * members of the cluster that are online. In this case we can potentially not send
- * the packets out (depending on sendWithNoMembers) and not cache them (for re-sending
- * if they where not ACK'ed).
- *
- */
- private int currentGroupSize = -1;
-
- /**
- * The last time a packet was sent from this node.
- */
- private long lastSendTime;
-
- /**
- * The max time we go without sending any packets.
- */
- private final int lastSendTimeFreqMillis;
-
- /**
- * The last time the cluster status was logged.
- */
- private long lastStatusTime = System.currentTimeMillis();
-
- /**
- * The max time we go before logging the cluster status.
- */
- private final int lastStatusTimeFreqMillis;
-
-
- private long totalTxnEventsSent;
- private long totalTxnEventsReceived;
-
- private long totalPacketsSent;
- private long totalBytesSent;
-
- private long totalPacketsResent;
- private long totalBytesResent;
-
- private long totalPacketsReceived;
- private long totalBytesReceived;
-
-
- public McastClusterManager(ContainerConfig containerConfig) {
-
- ContainerConfig.MulticastConfig config = containerConfig.getMulticastConfig();
-
- this.managerSleepMillis = config.getManagerSleepMillis();
- this.lastSendTimeFreqMillis = 1000 * config.getLastSendTimeFreqSecs();
- this.lastStatusTimeFreqMillis = 1000 * config.getLastStatusTimeFreqSecs();
-
- // the maximum number of times we will try to re-send a given packet before giving up sending
- this.maxResendOutgoing = config.getMaxResendOutgoingAttempts();
- // the maximum number of times we will ask for a packet to be resent to us before giving up asking
- int maxResendIncoming = config.getMaxResendIncomingRequests();
-
-
- int port = config.getListenPort();
- String addr = config.getListenAddress();
-
- int sendPort = config.getSendPort();
- String sendAddr = config.getSendAddress();
-
- // Sender options
- // Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507
- int maxSendPacketSize = config.getMaxSendPacketSize();
- // Whether to send packets even when there are no other members online
- this.sendWithNoMembers = config.isSendWithNoMembers();
-
- // Listener options
- // When multiple instances are on same box you need to broadcast back locally
- boolean disableLoopback = config.isDisableLoopback();
- int ttl = config.getListenTimeToLive();
- int timeout = config.getListenTimeout();
- int bufferSize = config.getListenBufferSize();
- // For multihomed environment the address the listener should bind to
- String mcastAddr = config.getListenBindAddress();
-
- InetAddress mcastBindAddress = null;
- if (mcastAddr != null) {
- try {
- mcastBindAddress = InetAddress.getByName(mcastAddr);
- } catch (UnknownHostException e) {
- String msg = "Error getting Multicast InetAddress for " + mcastAddr;
- throw new RuntimeException(msg, e);
- }
- }
-
- if (port == 0 || addr == null) {
- String msg = "One of these Multicast settings has not been set. " + "ebean.cluster.mcast.listen.port="
- + port + ", ebean.cluster.mcast.listen.address=" + addr;
-
- throw new IllegalArgumentException(msg);
- }
-
- this.managerThread = new Thread(this, "EbeanClusterMcastManager");
-
- this.packetWriter = new PacketWriter(maxSendPacketSize);
- this.localSender = new McastSender(port, addr, sendPort, sendAddr);
- this.localSenderHostPort = localSender.getSenderHostPort();
-
- this.packageControl = new McastPacketControl(this, localSenderHostPort, maxResendIncoming);
-
- this.listener = new McastListener(this, packageControl, port, addr, bufferSize, timeout, localSenderHostPort,
- disableLoopback, ttl, mcastBindAddress);
- }
-
-
- /**
- * The McastListener tells us there are no other members of the cluster that
- * are currently online.
- */
- protected void fromListenerTimeoutNoMembers() {
- synchronized (managerThread) {
- this.currentGroupSize = 0;
- }
- }
-
- /**
- * McastListener calls this method to get the manager to process messages.
- *
- * @param newMinAcked the minAcked packetId according to the listener
- * @param msgControl a control message to process
- * @param msgResend a Please re-send message to process
- * @param groupSize the number of other online members
- */
- protected void fromListener(long newMinAcked, MessageControl msgControl, MessageResend msgResend,
- int groupSize, long totalPacketsReceived, long totalBytesReceived,
- long totalTxnEventsReceived) {
-
- synchronized (managerThread) {
- if (newMinAcked > minAckedFromListener) {
- minAckedFromListener = newMinAcked;
- }
- if (msgControl != null) {
- controlMessages.add(msgControl);
- }
- if (msgResend != null) {
- resendMessages.add(msgResend);
- }
- // mostly interested when groupSize hits 0 (we are the only instance online).
- this.currentGroupSize = groupSize;
-
- // and some stats so we know how busy the listener has been
- this.totalPacketsReceived = totalPacketsReceived;
- this.totalBytesReceived = totalBytesReceived;
- this.totalTxnEventsReceived = totalTxnEventsReceived;
- }
- }
-
- /**
- * Get the overall status and activity of this cluster node.
- */
- public McastStatus getStatus() {
-
- synchronized (managerThread) {
- long currentPacketId = packetWriter.currentPacketId();
- String lastAcks = incomingPacketsLastAck.toString();
-
- return new McastStatus(currentGroupSize, outgoingPacketsCache.size(), currentPacketId, minAcked, lastAcks,
- totalTxnEventsSent, totalTxnEventsReceived, totalPacketsSent, totalPacketsResent,
- totalPacketsReceived,
- totalBytesSent, totalBytesResent, totalBytesReceived);
-
- }
- }
-
- /**
- * Periodically send out Ack, Re-send and Control messages.
- */
- public void run() {
-
- while (true) {
- try {
- // sleep for a little bit as we ACK packets periodically
- // rather than immediately. We will typically ACK many
- // messages from all cluster members in a single Packet
- Thread.sleep(managerSleepMillis);
-
- synchronized (managerThread) {
-
- handleControlMessages();
-
- handleResendMessages();
-
- if (currentGroupSize == 0) {
- // no members online so trim the entire outgoing packets cache
- int trimmedCount = outgoingPacketsCache.trimAll();
- if (trimmedCount > 0) {
- logger.debug("Cluster has no other members. Trimmed " + trimmedCount);
- }
-
- } else if (minAckedFromListener > minAcked) {
- // ACKs have come back so trim send packets cache
- outgoingPacketsCache.trimAcknowledgedMessages(minAckedFromListener);
- minAcked = minAckedFromListener;
- }
-
- // Get list of all the ACK messages required to sent since the last time.
- // This is effectively one ACK message per member of the cluster. The ACK
- // message covers all the packets received from the member up to
- // the gotAllPoint.
- // Also get any RESEND messages asking for packets that we have not
- // received between the gotAllPoint and the gotMaxPoint.
- AckResendMessages ackResendMessages = packageControl.getAckResendMessages(incomingPacketsLastAck);
-
- if (ackResendMessages.size() > 0) {
- // send the ACK and RESEND messages for all members of the
- // cluster typically in a single Packet
- if (sendMessages(false, ackResendMessages.getMessages())) {
- // update the last Ack position
- incomingPacketsLastAck.updateLastAck(ackResendMessages);
- }
- }
-
- if (lastSendTime < System.currentTimeMillis() - lastSendTimeFreqMillis) {
- // been quite for too long - send a Ping out
- sendPing();
- }
-
- if (lastStatusTimeFreqMillis > 0) {
- if (lastStatusTime < System.currentTimeMillis() - lastStatusTimeFreqMillis) {
- McastStatus status = getStatus();
- logger.info("Cluster Status: " + status.getSummary());
- lastStatusTime = System.currentTimeMillis();
- }
- }
-
- }
- } catch (Exception e) {
- logger.error("Error with Cluster Mcast Manager thread", e);
- }
- }
- }
-
- /**
- * We have been asked to Re-send some packets.
- */
- private void handleResendMessages() {
-
- if (resendMessages.size() > 0) {
-
- TreeSet s = new TreeSet();
- for (int i = 0; i < resendMessages.size(); i++) {
- MessageResend resendMsg = resendMessages.get(i);
- s.addAll(resendMsg.getResendPacketIds());
- }
-
- totalPacketsResent += s.size();
-
- for (Long resendPacketId : s) {
- Packet packet = outgoingPacketsCache.getPacket(resendPacketId);
- if (packet == null) {
- String msg = "Cluster unable to resend packet[" + resendPacketId + "] as it is no longer in the " +
- "outgoingPacketsCache";
- logger.error(msg);
- } else {
- int resendCount = packet.incrementResendCount();
- if (resendCount <= maxResendOutgoing) {
- resendPacket(packet);
- } else {
- String msg = "Cluster maxResendOutgoing [" + maxResendOutgoing + "] hit for packet " + resendPacketId
- + ". We will not try to send it anymore, removing it from the outgoingPacketsCache.";
- logger.error(msg);
- outgoingPacketsCache.remove(packet);
- }
- }
- }
- }
- }
-
- /**
- * Re-send a packet that a member didn't seem to receive.
- */
- private void resendPacket(Packet packet) {
- try {
- ++totalPacketsResent;
- totalBytesResent += localSender.sendPacket(packet);
- } catch (IOException e) {
- String msg = "Error trying to resend packet " + packet.getPacketId();
- logger.error(msg, e);
- }
- }
-
- /**
- * Handle Control messages (Join, Leave, Ping).
- */
- private void handleControlMessages() {
-
- boolean pingReponse = false;
- boolean joinReponse = false;
-
- for (int i = 0; i < controlMessages.size(); i++) {
- MessageControl message = controlMessages.get(i);
-
- short type = message.getControlType();
- switch (type) {
- case MessageControl.TYPE_JOIN:
- // a new member online, send back a Join Response
- logger.info("Cluster member Joined [" + message.getFromHostPort() + "]");
- joinReponse = true;
- break;
-
- case MessageControl.TYPE_JOINRESPONSE:
- logger.info("Cluster member Online [" + message.getFromHostPort() + "]");
- // do nothing
- break;
-
- case MessageControl.TYPE_PING:
- pingReponse = true;
- break;
-
- case MessageControl.TYPE_PINGRESPONSE:
- // do nothing
- break;
-
- case MessageControl.TYPE_LEAVE:
- // remove member. If/When that member comes back its
- // packetIds will have been reset
- incomingPacketsLastAck.remove(message.getFromHostPort());
- break;
-
- default:
- break;
- }
- }
- controlMessages.clear();
-
- if (joinReponse) {
- sendJoinResponse();
- }
- if (pingReponse) {
- sendPingResponse();
- }
- }
-
- /**
- * Say 'Leaving' and shutdown.
- */
- public void shutdown() {
- sendLeave();
- listener.shutdown();
- }
-
- /**
- * Startup listeners and 'Join'.
- */
- public void startup(ClusterManager clusterManager) {
- this.clusterManager = clusterManager;
- listener.startListening();
-
- this.managerThread.setDaemon(true);
- this.managerThread.start();
-
- sendJoin();
- }
-
- protected SpiEbeanServer getEbeanServer(String serverName) {
- return (SpiEbeanServer) clusterManager.getServer(serverName);
- }
-
- private void sendJoin() {
- sendControlMessage(true, MessageControl.TYPE_JOIN);
- }
-
- private void sendLeave() {
- sendControlMessage(false, MessageControl.TYPE_LEAVE);
- }
-
- private void sendJoinResponse() {
- sendControlMessage(true, MessageControl.TYPE_JOINRESPONSE);
- }
-
- private void sendPingResponse() {
- sendControlMessage(true, MessageControl.TYPE_PINGRESPONSE);
- }
-
- private void sendPing() {
- sendControlMessage(true, MessageControl.TYPE_PING);
- }
-
- private void sendControlMessage(boolean requiresAck, short controlType) {
- sendMessage(requiresAck, new MessageControl(controlType, localSenderHostPort));
- }
-
- private void sendMessage(boolean requiresAck, Message msg) {
- ArrayList messages = new ArrayList(1);
- messages.add(msg);
- sendMessages(requiresAck, messages);
- }
-
- private boolean sendMessages(boolean requiresAck, List messages) {
-
- synchronized (managerThread) {
- try {
-
- List packets = packetWriter.write(requiresAck, messages);
- sendPackets(requiresAck, packets);
- return true;
-
- } catch (IOException e) {
- String msg = "Error sending Messages " + messages;
- logger.error(msg, e);
- return false;
- }
- }
- }
-
- private boolean sendPackets(boolean requiresAck, List packets) throws IOException {
- if (currentGroupSize == 0 && !sendWithNoMembers) {
- // no other members online so not sending packets
- return false;
-
- } else {
- if (requiresAck) {
- // cache them until they have been ACK'ed
- outgoingPacketsCache.registerPackets(packets);
- }
- totalPacketsSent += packets.size();
- totalBytesSent += localSender.sendPackets(packets);
- lastSendTime = System.currentTimeMillis();
- return true;
- }
- }
-
- /**
- * Send the remoteTransEvent to all the other members of the cluster.
- */
- public void broadcast(RemoteTransactionEvent remoteTransEvent) {
-
- synchronized (managerThread) {
- try {
- List packets = packetWriter.write(remoteTransEvent);
- if (sendPackets(true, packets)) {
- ++totalTxnEventsSent;
- }
- } catch (IOException e) {
- String msg = "Error sending RemoteTransactionEvent " + remoteTransEvent;
- logger.error(msg, e);
- }
- }
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java
deleted file mode 100644
index 2eaaad97e..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java
+++ /dev/null
@@ -1,237 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.api.SpiEbeanServer;
-import com.avaje.ebeaninternal.server.cluster.Packet;
-import com.avaje.ebeaninternal.server.cluster.PacketTransactionEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.net.DatagramPacket;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.MulticastSocket;
-
-/**
- * Listens for Incoming packets.
- */
-public class McastListener implements Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(McastListener.class);
-
- private final McastClusterManager owner;
-
- private final McastPacketControl packetControl;
-
- private final MulticastSocket sock;
-
- private final Thread listenerThread;
-
- private final String localSenderHostPort;
-
- private final InetAddress group;
-
- private DatagramPacket pack;
-
- private final byte[] receiveBuffer;
-
- private volatile boolean shutdown;
- private volatile boolean shutdownComplete;
-
- private long totalPacketsReceived;
- private long totalBytesReceived;
- private long totalTxnEventsReceived;
-
- public McastListener(McastClusterManager owner, McastPacketControl packetControl, int port, String address,
- int bufferSize, int timeout, String localSenderHostPort,
- boolean disableLoopback, int ttl, InetAddress mcastBindAddress) {
-
- this.owner = owner;
- this.packetControl = packetControl;
- this.localSenderHostPort = localSenderHostPort;
- this.receiveBuffer = new byte[bufferSize];
- this.listenerThread = new Thread(this, "EbeanClusterMcastListener");
-
- String msg = "Cluster Multicast Listening address[" + address + "] port[" + port + "] disableLoopback[" + disableLoopback + "]";
- if (ttl >= 0) {
- msg += " ttl[" + ttl + "]";
- }
- if (mcastBindAddress != null) {
- msg += " mcastBindAddress[" + mcastBindAddress + "]";
- }
- logger.info(msg);
-
- try {
- this.group = InetAddress.getByName(address);
- this.sock = new MulticastSocket(port);
- this.sock.setSoTimeout(timeout);
-
- if (disableLoopback) {
- sock.setLoopbackMode(true);
- }
-
- if (mcastBindAddress != null) {
- // bind to a specific interface
- sock.setInterface(mcastBindAddress);
- }
-
- if (ttl >= 0) {
- sock.setTimeToLive(ttl);
- }
- sock.setReuseAddress(true);
- pack = new DatagramPacket(receiveBuffer, receiveBuffer.length);
- sock.joinGroup(group);
-
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public void startListening() {
- this.listenerThread.setDaemon(true);
- this.listenerThread.start();
-
- logger.info("Cluster Multicast Listener up and joined Group");
- }
-
- /**
- * Shutdown this listener.
- */
- public void shutdown() {
-
- shutdown = true;
- synchronized (listenerThread) {
- try {
- // wait max 20 seconds
- listenerThread.wait(20000);
- } catch (InterruptedException e) {
- logger.info("InterruptedException:" + e);
- }
- }
-
- if (!shutdownComplete) {
- String msg = "WARNING: Shutdown of McastListener did not complete?";
- System.err.println(msg);
- logger.warn(msg);
- }
-
- try {
- sock.leaveGroup(group);
- } catch (IOException e) {
- // send to syserr in case logging already shutdown
- e.printStackTrace();
- String msg = "Error leaving Multicast group";
- logger.info(msg, e);
- }
- try {
- sock.close();
- } catch (Exception e) {
- // send to syserr in case logging already shutdown
- e.printStackTrace();
- String msg = "Error closing Multicast socket";
- logger.info(msg, e);
- }
- }
-
- public void run() {
- while (!shutdown) {
- try {
- pack.setLength(receiveBuffer.length);
- sock.receive(pack);
-
- InetSocketAddress senderAddr = (InetSocketAddress) pack.getSocketAddress();
-
- String senderHostPort = senderAddr.getAddress().getHostAddress() + ":" + senderAddr.getPort();
-
- if (senderHostPort.equals(localSenderHostPort)) {
- if (logger.isTraceEnabled()) {
- logger.info("Ignoring message as sent by localSender: " + localSenderHostPort);
- }
- } else {
-
- byte[] data = pack.getData();
-
-
- ByteArrayInputStream bi = new ByteArrayInputStream(data);
- DataInputStream dataInput = new DataInputStream(bi);
-
- ++totalPacketsReceived;
- totalBytesReceived += pack.getLength();
-
- Packet header = Packet.readHeader(dataInput);
-
- long packetId = header.getPacketId();
- boolean ackMsg = packetId == 0;
-
- boolean processThisPacket = ackMsg || packetControl.isProcessPacket(senderHostPort, header.getPacketId());
-
- if (!processThisPacket) {
- if (logger.isTraceEnabled()) {
- logger.info("Already processed packet: " + header.getPacketId() + " type:" + header.getPacketType() + " len:" + data.length);
- }
- } else {
- if (logger.isTraceEnabled()) {
- logger.info("Incoming packet:" + header.getPacketId() + " type:" + header.getPacketType() + " len:" + data.length);
- }
- processPacket(senderHostPort, header, dataInput);
- }
- }
-
- } catch (java.net.SocketTimeoutException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("timeout", e);
- }
- packetControl.onListenerTimeout();
-
- } catch (IOException e) {
- logger.info("error ?", e);
- }
- }
-
- shutdownComplete = true;
-
- synchronized (listenerThread) {
- listenerThread.notifyAll();
- }
- }
-
- protected void processPacket(String senderHostPort, Packet header, DataInput dataInput) {
- try {
- switch (header.getPacketType()) {
- case Packet.TYPE_MESSAGES:
- packetControl.processMessagesPacket(senderHostPort, header, dataInput,
- totalPacketsReceived, totalBytesReceived, totalTxnEventsReceived);
- break;
-
- case Packet.TYPE_TRANSEVENT:
- ++totalTxnEventsReceived;
- processTransactionEventPacket(header, dataInput);
- break;
-
- default:
- String msg = "Unknown Packet type:" + header.getPacketType();
- logger.error(msg);
- break;
- }
- } catch (IOException e) {
- // need to ask to get this packet resent...
- String msg = "Error reading Packet " + header.getPacketId() + " type:" + header.getPacketType();
- logger.error(msg, e);
- }
- }
-
- private void processTransactionEventPacket(Packet header, DataInput dataInput) throws IOException {
-
- SpiEbeanServer server = owner.getEbeanServer(header.getServerName());
-
- PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server);
- tranEventPacket.read(dataInput);
-
- server.remoteTransactionEvent(tranEventPacket.getEvent());
- }
-
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java
deleted file mode 100644
index 6c456b4ea..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java
+++ /dev/null
@@ -1,125 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.Packet;
-import com.avaje.ebeaninternal.server.cluster.PacketMessages;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.DataInput;
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.List;
-
-/**
- * Helps co-ordinate Packet information between the McastListener and the
- * McastClusterManager.
- *
- * @author rbygrave
- */
-public class McastPacketControl {
-
- private static final Logger logger = LoggerFactory.getLogger(McastPacketControl.class);
-
- private final String localSenderHostPort;
-
- private final McastClusterManager owner;
-
- private final HashSet groupMembers = new HashSet();
-
- private final OutgoingPacketsAcked outgoingPacketsAcked = new OutgoingPacketsAcked();
-
- private final IncomingPacketsProcessed incomingPacketsProcessed;
-
- public McastPacketControl(McastClusterManager owner, String localSenderHostPort, int maxResendIncoming) {
- this.owner = owner;
- this.localSenderHostPort = localSenderHostPort;
- this.incomingPacketsProcessed = new IncomingPacketsProcessed(maxResendIncoming);
- }
-
- /**
- * Handle special case where cluster doesn't have any members and we don't
- * get any responses. Need to tell the sender side that the group size is 0.
- */
- protected void onListenerTimeout() {
- if (groupMembers.size() == 0) {
- owner.fromListenerTimeoutNoMembers();
- }
- }
-
- protected void processMessagesPacket(String senderHostPort, Packet header, DataInput dataInput,
- long totalPacketsReceived, long totalBytesReceived, long totalTransEventsReceived) throws IOException {
-
- PacketMessages packetMessages = PacketMessages.forRead(header);
- packetMessages.read(dataInput);
- List messages = packetMessages.getMessages();
-
- if (logger.isTraceEnabled()) {
- logger.trace("INCOMING Messages " + messages);
- }
- // messages are for all nodes in the cluster so
- // we need to filter looking for messages pertaining
- // to this (senderHostPort)
-
- MessageControl control = null;
- MessageAck ack = null;
- MessageResend resend = null;
-
- // filter for relevant messages to this node
- for (int i = 0; i < messages.size(); i++) {
- Message message = messages.get(i);
- if (message.isControlMessage()) {
- // any 'control' message is interesting
- control = (MessageControl) message;
-
- } else if (localSenderHostPort.equals(message.getToHostPort())) {
- if (message instanceof MessageAck) {
- ack = (MessageAck) message;
- } else if (message instanceof MessageResend) {
- resend = (MessageResend) message;
- } else {
- logger.error("Expecting a MessageAck or MessageResend but got a "
- + message.getClass().getName());
- }
- }
- }
-
- if (control != null) {
- if (control.getControlType() == MessageControl.TYPE_LEAVE) {
- groupMembers.remove(senderHostPort);
- logger.info("Cluster member leaving [" + senderHostPort + "] " + groupMembers.size()
- + " other members left");
- outgoingPacketsAcked.removeMember(senderHostPort);
- incomingPacketsProcessed.removeMember(senderHostPort);
- } else {
- groupMembers.add(senderHostPort);
- }
- }
-
- long newMin = 0;
- if (ack != null) {
- newMin = outgoingPacketsAcked.receivedAck(senderHostPort, ack);
- }
-
- if (newMin > 0 || control != null || resend != null) {
- int groupSize = groupMembers.size();
- // synchronised on the managerThread
- owner.fromListener(newMin, control, resend, groupSize,
- totalPacketsReceived, totalBytesReceived, totalTransEventsReceived);
- }
- }
-
- /**
- * Return true if we should process this packet. Return false if we have
- * already processed the packet.
- */
- public boolean isProcessPacket(String memberKey, long packetId) {
-
- return incomingPacketsProcessed.isProcessPacket(memberKey, packetId);
- }
-
- public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
-
- return incomingPacketsProcessed.getAckResendMessages(lastAck);
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java
deleted file mode 100644
index 80994cd8d..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java
+++ /dev/null
@@ -1,114 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.Packet;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.DatagramPacket;
-import java.net.DatagramSocket;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.List;
-
-/**
- * Handles the sending of Packets via DatagramPacket.
- */
-public class McastSender {
-
- private static final Logger logger = LoggerFactory.getLogger(McastSender.class);
-
- private final int port;
-
- private final InetAddress inetAddress;
-
- private final DatagramSocket sock;
-
- private final InetSocketAddress sendAddr;
-
- private final String senderHostPort;
-
-
- public McastSender(int port, String address, int sendPort, String sendAddress) {
-
- try {
- this.port = port;
- this.inetAddress = InetAddress.getByName(address);
-
- InetAddress sendInetAddress;
- if (sendAddress != null) {
- sendInetAddress = InetAddress.getByName(sendAddress);
- } else {
- sendInetAddress = InetAddress.getLocalHost();
- }
-
- if (sendPort > 0) {
- this.sock = new DatagramSocket(sendPort, sendInetAddress);
- } else {
- this.sock = new DatagramSocket(new InetSocketAddress(sendInetAddress, 0));
- }
-
- String msg = "Cluster Multicast Sender on[" + sendInetAddress.getHostAddress() + ":" + sock.getLocalPort() + "]";
- logger.info(msg);
-
- this.sendAddr = new InetSocketAddress(sendInetAddress, sock.getLocalPort());
- this.senderHostPort = sendInetAddress.getHostAddress() + ":" + sock.getLocalPort();
-
- } catch (Exception e) {
- String msg = "McastSender port:" + port + " sendPort:" + sendPort + " " + address;
- throw new RuntimeException(msg, e);
- }
- }
-
- /**
- * Return the send Address so that if we have loopback messages we can
- * detect if they where sent by this local sender and hence should be
- * ignored.
- */
- public InetSocketAddress getAddress() {
- return sendAddr;
- }
-
- /**
- * Return the Host and Port of the sender. This is used to uniquely identify
- * this instance in the cluster.
- */
- public String getSenderHostPort() {
- return senderHostPort;
- }
-
- /**
- * Send the packet.
- */
- public int sendPacket(Packet packet) throws IOException {
-
- byte[] pktBytes = packet.getBytes();
-
- if (logger.isDebugEnabled()) {
- logger.debug("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length);
- }
-
- if (pktBytes.length > 65507) {
- logger.warn("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length
- + " likely to be truncated using UDP with a MAXIMUM length of 65507");
- }
-
- DatagramPacket pack = new DatagramPacket(pktBytes, pktBytes.length, inetAddress, port);
- sock.send(pack);
-
- return pktBytes.length;
- }
-
- /**
- * Send the list of Packets.
- */
- public int sendPackets(List packets) throws IOException {
-
- int totalBytes = 0;
- for (int i = 0; i < packets.size(); i++) {
- totalBytes += sendPacket(packets.get(i));
- }
- return totalBytes;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java
deleted file mode 100644
index e4c1659cf..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java
+++ /dev/null
@@ -1,134 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-/**
- * Gives an overall status of this Cluster instance.
- *
- * Ideally you want to see relatively low Re-send statistics.
- *
- */
-public class McastStatus {
-
- private final long totalTxnEventsSent;
- private final long totalTxnEventsReceived;
-
- private final long totalPacketsSent;
- private final long totalPacketsResent;
- private final long totalPacketsReceived;
-
- private final long totalBytesSent;
- private final long totalBytesResent;
- private final long totalBytesReceived;
-
- private final int currentGroupSize;
- private final int outgoingPacketsCacheSize;
-
- private final long currentPacketId;
- private final long minAckedPacketId;
- private final String lastOutgoingAcks;
-
- public String getSummary() {
-
- //noinspection StringBufferReplaceableByString
- StringBuilder sb = new StringBuilder(80);
- sb.append("txnOut:").append(totalTxnEventsSent).append("; ");
- sb.append("txnIn:").append(totalTxnEventsReceived).append("; ");
- sb.append("outPackets:").append(totalPacketsSent).append("; ");
- sb.append("outBytes:").append(totalBytesSent).append("; ");
- sb.append("inPackets:").append(totalPacketsReceived).append("; ");
- sb.append("inBytes:").append(totalBytesReceived).append("; ");
- sb.append("resentPackets:").append(totalPacketsResent).append("; ");
- sb.append("resentBytes:").append(totalBytesResent).append("; ");
- sb.append("groupSize:").append(currentGroupSize).append("; ");
- sb.append("cache:").append(outgoingPacketsCacheSize).append("; ");
- sb.append("currentPacket:").append(currentPacketId).append("; ");
- sb.append("minAckedPacket:").append(minAckedPacketId).append("; ");
- sb.append("lastAck:").append(lastOutgoingAcks).append("; ");
-
- return sb.toString();
- }
-
- public McastStatus(int currentGroupSize,
- int outgoingPacketsCacheSize,
- long currentPacketId,
- long minAckedPacketId,
- String lastOutgoingAcks,
- long totalTransEventsSent,
- long totalTransEventsReceived,
- long totalPacketsSent,
- long totalPacketsResent,
- long totalPacketsReceived,
- long totalBytesSent,
- long totalBytesResent,
- long totalBytesReceived) {
-
- this.currentGroupSize = currentGroupSize;
- this.outgoingPacketsCacheSize = outgoingPacketsCacheSize;
- this.currentPacketId = currentPacketId;
- this.minAckedPacketId = minAckedPacketId;
- this.lastOutgoingAcks = lastOutgoingAcks;
- this.totalTxnEventsSent = totalTransEventsSent;
- this.totalTxnEventsReceived = totalTransEventsReceived;
- this.totalPacketsSent = totalPacketsSent;
- this.totalPacketsResent = totalPacketsResent;
- this.totalPacketsReceived = totalPacketsReceived;
-
- this.totalBytesSent = totalBytesSent;
- this.totalBytesResent = totalBytesResent;
- this.totalBytesReceived = totalBytesReceived;
-
- }
-
-
- public long getTotalTxnEventsReceived() {
- return totalTxnEventsReceived;
- }
-
- public long getTotalPacketsReceived() {
- return totalPacketsReceived;
- }
-
- public long getTotalBytesSent() {
- return totalBytesSent;
- }
-
- public long getTotalBytesResent() {
- return totalBytesResent;
- }
-
- public long getTotalBytesReceived() {
- return totalBytesReceived;
- }
-
- public String getLastOutgoingAcks() {
- return lastOutgoingAcks;
- }
-
- public int getOutgoingPacketsCacheSize() {
- return outgoingPacketsCacheSize;
- }
-
- public long getCurrentPacketId() {
- return currentPacketId;
- }
-
- public long getMinAckedPacketId() {
- return minAckedPacketId;
- }
-
- public long getTotalTxnEventsSent() {
- return totalTxnEventsSent;
- }
-
- public long getTotalPacketsSent() {
- return totalPacketsSent;
- }
-
- public long getTotalPacketsResent() {
- return totalPacketsResent;
- }
-
- public long getCurrentGroupSize() {
- return currentGroupSize;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java
deleted file mode 100644
index 7f21a7cf8..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
-
-import java.io.IOException;
-
-public interface Message {
-
- void writeBinaryMessage(BinaryMessageList msgList) throws IOException;
-
- boolean isControlMessage();
-
- String getToHostPort();
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java
deleted file mode 100644
index a261a4060..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java
+++ /dev/null
@@ -1,57 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
-import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
-
-import java.io.DataInput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-public class MessageAck implements Message {
-
- private final String toHostPort;
-
- private final long gotAllPacketId;
-
- public MessageAck(String toHostPort, long gotAllPacketId) {
- this.toHostPort = toHostPort;
- this.gotAllPacketId = gotAllPacketId;
- }
-
- public String toString() {
- return "Ack " + toHostPort + " " + gotAllPacketId;
- }
-
- public boolean isControlMessage() {
- return false;
- }
-
- public String getToHostPort() {
- return toHostPort;
- }
-
- public long getGotAllPacketId() {
- return gotAllPacketId;
- }
-
-
- public static MessageAck readBinaryMessage(DataInput dataInput) throws IOException {
-
- String hostPort = dataInput.readUTF();
- long gotAllPacketId = dataInput.readLong();
- return new MessageAck(hostPort, gotAllPacketId);
- }
-
- public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
-
- BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
-
- DataOutputStream os = m.getOs();
- os.writeInt(BinaryMessage.TYPE_MSGACK);
- os.writeUTF(toHostPort);
- os.writeLong(gotAllPacketId);
- os.flush();
-
- msgList.add(m);
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java
deleted file mode 100644
index e917c9db8..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java
+++ /dev/null
@@ -1,77 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
-import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
-
-import java.io.DataInput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-public class MessageControl implements Message {
-
- public static final short TYPE_JOIN = 1;
- public static final short TYPE_LEAVE = 2;
- public static final short TYPE_PING = 3;
- public static final short TYPE_JOINRESPONSE = 7;
- public static final short TYPE_PINGRESPONSE = 8;
-
- private final short controlType;
- private final String fromHostPort;
-
- public static MessageControl readBinaryMessage(DataInput dataInput) throws IOException {
- short controlType = dataInput.readShort();
- String hostPort = dataInput.readUTF();
- return new MessageControl(controlType, hostPort);
- }
-
- public MessageControl(short controlType, String helloFromHostPort) {
- this.controlType = controlType;
- this.fromHostPort = helloFromHostPort;
- }
-
-
- public String toString() {
- switch (controlType) {
- case TYPE_JOIN:
- return "Join " + fromHostPort;
- case TYPE_LEAVE:
- return "Leave " + fromHostPort;
- case TYPE_PING:
- return "Ping " + fromHostPort;
- case TYPE_PINGRESPONSE:
- return "PingResponse " + fromHostPort;
-
- default:
- throw new RuntimeException("Invalid controlType " + controlType);
- }
- }
-
- public boolean isControlMessage() {
- return true;
- }
-
- public short getControlType() {
- return controlType;
- }
-
- public String getToHostPort() {
- return "*";
- }
-
- public String getFromHostPort() {
- return fromHostPort;
- }
-
- public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
-
- BinaryMessage m = new BinaryMessage(fromHostPort.length() * 2 + 10);
-
- DataOutputStream os = m.getOs();
- os.writeInt(BinaryMessage.TYPE_MSGCONTROL);
- os.writeShort(controlType);
- os.writeUTF(fromHostPort);
- os.flush();
-
- msgList.add(m);
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java
deleted file mode 100644
index c712b6c67..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java
+++ /dev/null
@@ -1,77 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
-import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
-
-import java.io.DataInput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-public class MessageResend implements Message {
-
- private final String toHostPort;
-
- private final List resendPacketIds;
-
- public MessageResend(String toHostPort, List resendPacketIds) {
- this.toHostPort = toHostPort;
- this.resendPacketIds = resendPacketIds;
- }
-
- public MessageResend(String toHostPort) {
- this(toHostPort, new ArrayList(4));
- }
-
- public String toString() {
- return "Resend " + toHostPort + " " + resendPacketIds;
- }
-
- public boolean isControlMessage() {
- return false;
- }
-
- public String getToHostPort() {
- return toHostPort;
- }
-
- public void add(long packetId) {
- resendPacketIds.add(packetId);
- }
-
- public List getResendPacketIds() {
- return resendPacketIds;
- }
-
- public static MessageResend readBinaryMessage(DataInput dataInput) throws IOException {
-
- String hostPort = dataInput.readUTF();
-
- MessageResend msg = new MessageResend(hostPort);
-
- int numberOfPacketIds = dataInput.readInt();
- for (int i = 0; i < numberOfPacketIds; i++) {
- long packetId = dataInput.readLong();
- msg.add(packetId);
- }
-
- return msg;
- }
-
- public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
-
- BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
-
- DataOutputStream os = m.getOs();
- os.writeInt(BinaryMessage.TYPE_MSGRESEND);
- os.writeUTF(toHostPort);
- os.writeInt(resendPacketIds.size());
- for (int i = 0; i < resendPacketIds.size(); i++) {
- Long packetId = resendPacketIds.get(i);
- os.writeLong(packetId.longValue());
- }
- os.flush();
- msgList.add(m);
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java
deleted file mode 100644
index 8e7a836e9..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java
+++ /dev/null
@@ -1,94 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class OutgoingPacketsAcked {
-
- private long minimumGotAllPacketId;
-
- private final Map recievedByMap = new HashMap();
-
- public int getGroupSize() {
- synchronized (this) {
- return recievedByMap.size();
- }
- }
-
- public long getMinimumGotAllPacketId() {
- synchronized (this) {
- return minimumGotAllPacketId;
- }
- }
-
- public void removeMember(String groupMember) {
- synchronized (this) {
- recievedByMap.remove(groupMember);
- resetGotAllMin();
- }
- }
-
- private boolean resetGotAllMin() {
-
- long tempMin = Long.MAX_VALUE;
-
- for (GroupMemberAck groupMemAck : recievedByMap.values()) {
- long memberMin = groupMemAck.getGotAllPacketId();
- if (memberMin < tempMin) {
- tempMin = memberMin;
- }
- }
-
- if (tempMin != minimumGotAllPacketId) {
- minimumGotAllPacketId = tempMin;
- return true;
- } else {
- return false;
- }
- }
-
- public long receivedAck(String groupMember, MessageAck ack) {
-
- synchronized (this) {
-
- boolean checkMin;
-
- GroupMemberAck groupMemberAck = recievedByMap.get(groupMember);
- if (groupMemberAck == null) {
- groupMemberAck = new GroupMemberAck();
- groupMemberAck.setIfBigger(ack.getGotAllPacketId());
- recievedByMap.put(groupMember, groupMemberAck);
- checkMin = true;
- } else {
- checkMin = groupMemberAck.getGotAllPacketId() == minimumGotAllPacketId;
- groupMemberAck.setIfBigger(ack.getGotAllPacketId());
- }
-
- boolean minChanged = false;
-
- if (checkMin || minimumGotAllPacketId == 0) {
- minChanged = resetGotAllMin();
- }
-
- return minChanged ? minimumGotAllPacketId : 0;
- }
- }
-
- private static class GroupMemberAck {
-
- private long gotAllPacketId;
-
- private GroupMemberAck() {
- }
-
- private long getGotAllPacketId() {
- return gotAllPacketId;
- }
-
- private void setIfBigger(long newGotAll) {
- if (newGotAll > gotAllPacketId) {
- gotAllPacketId = newGotAll;
- }
- }
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java
deleted file mode 100644
index 6ffaac2f2..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java
+++ /dev/null
@@ -1,63 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.mcast;
-
-import com.avaje.ebeaninternal.server.cluster.Packet;
-
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
-
-/**
- * Cache of the outgoing packets.
- *
- * These are held until we receive ACKs from the other members of the cluster to
- * say they have received the packets.
- *
- */
-public class OutgoingPacketsCache {
-
- private final Map packetMap = new TreeMap();
-
- public int size() {
- return packetMap.size();
- }
-
- public Packet getPacket(Long packetId) {
- return packetMap.get(packetId);
- }
-
- public String toString() {
- return packetMap.keySet().toString();
- }
-
- /**
- * Remove the packet when we give up trying to send it out.
- */
- public void remove(Packet packet) {
- packetMap.remove(packet.getPacketId());
- }
-
- public void registerPackets(List packets) {
- for (int i = 0; i < packets.size(); i++) {
- Packet p = packets.get(i);
- packetMap.put(p.getPacketId(), p);
- }
- }
-
- public int trimAll() {
- int size = packetMap.size();
- packetMap.clear();
- return size;
- }
-
- public void trimAcknowledgedMessages(long minAcked) {
- Iterator it = packetMap.keySet().iterator();
- while (it.hasNext()) {
- Long pktId = it.next();
- if (minAcked >= pktId) {
- it.remove();
- }
- }
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java
deleted file mode 100644
index a4c89516d..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.Socket;
-
-/**
- * This parses and dispatches a request to the appropriate handler.
- *
- * Looks up the appropriate RequestHandler
- * and then gets it to process the Client request.
- *
- * Note that this is a Runnable because it is assigned to the ThreadPool.
- */
-class RequestProcessor implements Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(RequestProcessor.class);
-
- private final Socket clientSocket;
-
- private final SocketClusterBroadcast owner;
-
- private final String hostPort;
-
- /**
- * Create including the Listener (used to lookup the Request Handler) and
- * the socket itself.
- */
- public RequestProcessor(SocketClusterBroadcast owner, Socket clientSocket) {
- this.clientSocket = clientSocket;
- this.owner = owner;
- this.hostPort = owner.getHostPort();
- }
-
- /**
- * This will parse out the command. Lookup the appropriate Handler and
- * pass the information to the handler for processing.
- *
Dev Note: the command parsing is processed here so that it is preformed
- * by the assigned thread rather than the listeners thread.
- */
- public void run() {
- try {
- logger.trace("start listening for cluster messages");
- SocketConnection sc = new SocketConnection(clientSocket);
- while (true) {
- if (owner.process(sc)) {
- // got the offline message or timeout
- break;
- }
- }
- logger.trace("disconnecting: {}", hostPort);
- sc.disconnect();
-
- } catch (Exception e) {
- logger.error("Error listening for messages - " + owner.getHostPort(), e);
- }
- }
-
-}
\ No newline at end of file
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java
deleted file mode 100644
index acefd636d..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java
+++ /dev/null
@@ -1,132 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.ObjectOutputStream;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-
-
-/**
- * The client side of the socket clustering.
- */
-class SocketClient {
-
- private static final Logger logger = LoggerFactory.getLogger(SocketClient.class);
-
- private final InetSocketAddress address;
-
- private final String hostPort;
-
- private boolean online;
-
- private Socket socket;
- private OutputStream os;
- private ObjectOutputStream oos;
-
- /**
- * Construct with an IP address and port.
- */
- public SocketClient(InetSocketAddress address) {
- this.address = address;
- this.hostPort = address.getHostName() + ":" + address.getPort();
- }
-
- public String toString() {
- return address.toString();
- }
-
- public String getHostPort() {
- return hostPort;
- }
-
- public int getPort() {
- return address.getPort();
- }
-
- public boolean isOnline() {
- return online;
- }
-
- public void setOnline(boolean online) throws IOException {
- if (online) {
- setOnline();
- } else {
- disconnect();
- }
- }
-
-
- /**
- * Set whether the client is thought to be online.
- */
- private void setOnline() throws IOException {
- connect();
- this.online = true;
- }
-
- public void reconnect() throws IOException {
- disconnect();
- connect();
- }
-
- private void connect() throws IOException {
- if (socket != null) {
- throw new IllegalStateException("Already got a socket connection?");
- }
- Socket s = new Socket();
- s.setKeepAlive(true);
- s.connect(address);
-
- this.socket = s;
- this.os = socket.getOutputStream();
- }
-
- public void disconnect() {
- this.online = false;
- if (socket != null) {
-
- try {
- socket.close();
- } catch (IOException e) {
- String msg = "Error disconnecting from Cluster member " + hostPort;
- logger.info(msg, e);
- }
-
- os = null;
- oos = null;
- socket = null;
- }
- }
-
- public boolean register(SocketClusterMessage registerMsg) {
-
- try {
- setOnline();
- send(registerMsg);
- return true;
- } catch (IOException e) {
- disconnect();
- return false;
- }
- }
-
- public void send(SocketClusterMessage msg) throws IOException {
-
- if (online) {
- writeObject(msg);
- }
- }
-
- private void writeObject(Object object) throws IOException {
- if (oos == null) {
- this.oos = new ObjectOutputStream(os);
- }
- oos.writeObject(object);
- oos.flush();
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java
deleted file mode 100644
index d77878b50..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java
+++ /dev/null
@@ -1,250 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import com.avaje.ebean.config.ContainerConfig;
-import com.avaje.ebeaninternal.api.SpiEbeanServer;
-import com.avaje.ebeaninternal.server.cluster.ClusterBroadcast;
-import com.avaje.ebeaninternal.server.cluster.ClusterManager;
-import com.avaje.ebeaninternal.server.cluster.DataHolder;
-import com.avaje.ebeaninternal.server.cluster.SerialiseTransactionHelper;
-import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.net.InetSocketAddress;
-import java.util.HashMap;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * Broadcast messages across the cluster using sockets.
- */
-public class SocketClusterBroadcast implements ClusterBroadcast {
-
- private static final Logger logger = LoggerFactory.getLogger(SocketClusterBroadcast.class);
-
- private final SocketClient local;
-
- private final HashMap clientMap;
-
- private final SocketClusterListener listener;
-
- private final SocketClient[] members;
-
- private ClusterManager clusterManager;
-
- private final TxnSerialiseHelper txnSerialiseHelper = new TxnSerialiseHelper();
-
- private final AtomicInteger txnOutgoing = new AtomicInteger();
- private final AtomicInteger txnIncoming = new AtomicInteger();
-
- public SocketClusterBroadcast(ContainerConfig containerConfig) {
-
- ContainerConfig.SocketConfig socketConfig = containerConfig.getSocketConfig();
-
- String localHostPort = socketConfig.getLocalHostPort();
- List members = socketConfig.getMembers();
-
- logger.info("Clustering using Sockets local[" + localHostPort + "] members[" + members + "]");
-
- this.local = new SocketClient(parseFullName(localHostPort));
- this.clientMap = new HashMap();
-
- for (String memberHostPort : members) {
- InetSocketAddress member = parseFullName(memberHostPort);
- SocketClient client = new SocketClient(member);
- if (!local.getHostPort().equalsIgnoreCase(client.getHostPort())) {
- // don't add the local one ...
- clientMap.put(client.getHostPort(), client);
- }
- }
-
- this.members = clientMap.values().toArray(new SocketClient[clientMap.size()]);
- this.listener = new SocketClusterListener(this, local.getPort(), socketConfig.getCoreThreads(), socketConfig.getMaxThreads(), socketConfig.getThreadPoolName());
- }
-
- public String getHostPort() {
- return local.getHostPort();
- }
-
- /**
- * Return the current status of this instance.
- */
- public SocketClusterStatus getStatus() {
-
- // count of online members
- int currentGroupSize = 0;
- for (int i = 0; i < members.length; i++) {
- if (members[i].isOnline()) {
- ++currentGroupSize;
- }
- }
- int txnIn = txnIncoming.get();
- int txnOut = txnOutgoing.get();
-
- return new SocketClusterStatus(currentGroupSize, txnIn, txnOut);
- }
-
- public void startup(ClusterManager clusterManager) {
-
- this.clusterManager = clusterManager;
- listener.startListening();
- register();
- }
-
- public void shutdown() {
- deregister();
- listener.shutdown();
- }
-
- /**
- * Register with all the other members of the Cluster.
- */
- private void register() {
-
- SocketClusterMessage h = SocketClusterMessage.register(local.getHostPort(), true);
-
- for (int i = 0; i < members.length; i++) {
- boolean online = members[i].register(h);
- logger.info("Cluster Member [{}] online[{}]", members[i].getHostPort(), online);
- }
- }
-
- protected void setMemberOnline(String fullName, boolean online) throws IOException {
- synchronized (clientMap) {
- logger.info("Cluster Member [{}] online[{}]", fullName, online);
- SocketClient member = clientMap.get(fullName);
- member.setOnline(online);
- }
- }
-
- private void send(SocketClient client, SocketClusterMessage msg) {
-
- try {
- // alternative would be to connect/disconnect here but prefer to use keepalive
- if (logger.isTraceEnabled()) {
- logger.trace("... send to member {} broadcast msg: {}", client, msg);
- }
- client.send(msg);
-
- } catch (Exception ex) {
- logger.error("Error sending message", ex);
- try {
- client.reconnect();
- } catch (IOException e) {
- logger.error("Error trying to reconnect", ex);
- }
- }
- }
-
- /**
- * Send the payload to all the members of the cluster.
- */
- public void broadcast(RemoteTransactionEvent remoteTransEvent) {
- try {
- txnOutgoing.incrementAndGet();
- DataHolder dataHolder = txnSerialiseHelper.createDataHolder(remoteTransEvent);
- SocketClusterMessage msg = SocketClusterMessage.transEvent(dataHolder);
- broadcast(msg);
- } catch (Exception e) {
- logger.error("Error sending RemoteTransactionEvent " + remoteTransEvent + " to cluster members.", e);
- }
- }
-
- protected void broadcast(SocketClusterMessage msg) {
-
- if (logger.isTraceEnabled()) {
- logger.trace("... broadcast msg: " + msg);
- }
- for (int i = 0; i < members.length; i++) {
- send(members[i], msg);
- }
- }
-
- /**
- * Leave the cluster.
- */
- private void deregister() {
-
- SocketClusterMessage h = SocketClusterMessage.register(local.getHostPort(), false);
- broadcast(h);
- for (int i = 0; i < members.length; i++) {
- members[i].disconnect();
- }
- }
-
- /**
- * Process an incoming Cluster message.
- */
- protected boolean process(SocketConnection request) throws ClassNotFoundException {
-
- try {
- SocketClusterMessage h = (SocketClusterMessage) request.readObject();
- if (logger.isTraceEnabled()) {
- logger.trace("... received msg: {}", h);
- }
-
- if (h.isRegisterEvent()) {
- setMemberOnline(h.getRegisterHost(), h.isRegister());
-
- } else {
- txnIncoming.incrementAndGet();
- DataHolder dataHolder = h.getDataHolder();
- RemoteTransactionEvent transEvent = txnSerialiseHelper.read(dataHolder);
- transEvent.run();
- }
-
- // instance shutting down
- return h.isRegisterEvent() && !h.isRegister();
-
- } catch (InterruptedIOException e) {
- logger.info("Timeout waiting for message", e);
- try {
- request.disconnect();
- } catch (IOException ex) {
- logger.info("Error disconnecting after timeout", ex);
- }
- return true;
-
- } catch (EOFException e) {
- logger.info("EOF disconnecting");
- return true;
- } catch (IOException e) {
- logger.info("IO Error waiting/reading message", e);
- return true;
- }
- }
-
- /**
- * Parse a host:port into a InetSocketAddress.
- */
- private InetSocketAddress parseFullName(String hostAndPort) {
-
- try {
- hostAndPort = hostAndPort.trim();
- int colonPos = hostAndPort.indexOf(":");
- if (colonPos == -1) {
- String msg = "No colon \":\" in " + hostAndPort;
- throw new IllegalArgumentException(msg);
- }
- String host = hostAndPort.substring(0, colonPos);
- String sPort = hostAndPort.substring(colonPos + 1, hostAndPort.length());
- int port = Integer.parseInt(sPort);
-
- return new InetSocketAddress(host, port);
-
- } catch (Exception ex) {
- throw new RuntimeException("Error parsing [" + hostAndPort + "] for the form [host:port]", ex);
- }
- }
-
- class TxnSerialiseHelper extends SerialiseTransactionHelper {
-
- @Override
- public SpiEbeanServer getEbeanServer(String serverName) {
- return (SpiEbeanServer) clusterManager.getServer(serverName);
- }
- }
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java
deleted file mode 100644
index 10d55b978..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java
+++ /dev/null
@@ -1,143 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import com.avaje.ebeaninternal.server.lib.DaemonThreadPool;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.SocketException;
-
-
-/**
- * Serverside multithreaded socket listener. Accepts connections and dispatches
- * them to an appropriate handler.
- *
- * This is designed as a single port listener, where part of the connection
- * protocol determines which service the client is requesting (rather than a
- * port per service).
- *
- *
- * It has its own daemon background thread that handles the accept() loop on the
- * ServerSocket.
- *
- */
-class SocketClusterListener implements Runnable {
-
- private static final Logger logger = LoggerFactory.getLogger(SocketClusterListener.class);
-
- /**
- * The server socket used to listen for requests.
- */
- private final ServerSocket serverListenSocket;
-
- /**
- * The listening thread.
- */
- private final Thread listenerThread;
-
- /**
- * The pool of threads that actually do the parsing execution of requests.
- */
- private final DaemonThreadPool threadPool;
-
- private final SocketClusterBroadcast owner;
-
- /**
- * shutting down flag.
- */
- boolean doingShutdown;
-
- /**
- * Whether the listening thread is busy assigning a request to a thread.
- */
- boolean isActive;
-
- /**
- * Construct with a given thread pool name.
- */
- public SocketClusterListener(SocketClusterBroadcast owner, int port, int coreThreads, int maxThreads, String poolName) {
- this.owner = owner;
- this.threadPool = new DaemonThreadPool(coreThreads, maxThreads, 60, 30, poolName);
- try {
- this.serverListenSocket = new ServerSocket(port);
- this.serverListenSocket.setSoTimeout(60000);
- this.listenerThread = new Thread(this, "EbeanClusterListener");
-
- } catch (IOException e) {
- String msg = "Error starting cluster socket listener on port " + port;
- throw new RuntimeException(msg, e);
- }
- }
-
- /**
- * Start listening for requests.
- */
- public void startListening() {
- logger.trace("... startListening()");
- this.listenerThread.setDaemon(true);
- this.listenerThread.start();
- }
-
- /**
- * Shutdown this listener.
- */
- public void shutdown() {
- doingShutdown = true;
- try {
- if (isActive) {
- synchronized (listenerThread) {
- try {
- listenerThread.wait(1000);
- } catch (InterruptedException e) {
- // OK to ignore as expected to Interrupt for shutdown.
- }
- }
- }
- listenerThread.interrupt();
- serverListenSocket.close();
- } catch (IOException e) {
- logger.error("Error shutting down listener", e);
- }
-
- threadPool.shutdown();
- }
-
- /**
- * This is a runnable and so this must be public. Don't call this externally
- * but rather call the startListening() method.
- */
- public void run() {
- // run in loop until doingShutdown is true...
- while (!doingShutdown) {
- try {
- synchronized (listenerThread) {
- Socket clientSocket = serverListenSocket.accept();
- isActive = true;
-
- Runnable request = new RequestProcessor(owner, clientSocket);
- threadPool.execute(request);
-
- isActive = false;
- }
- } catch (SocketException e) {
- if (doingShutdown) {
- logger.info("doingShutdown and accept threw:" + e.getMessage());
- } else {
- logger.error("Error while listening", e);
- }
- } catch (InterruptedIOException e) {
- // this will happen when the server is very quiet.
- // that is, no requests
- logger.debug("Possibly expected due to accept timeout? {}", e.getMessage());
-
- } catch (IOException e) {
- // log it and continue in the loop...
- logger.error("IOException processing cluster message", e);
- }
- }
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java
deleted file mode 100644
index 07bfd6d11..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import com.avaje.ebeaninternal.server.cluster.DataHolder;
-import com.avaje.ebeaninternal.server.cluster.Packet;
-
-import java.io.Serializable;
-
-/**
- * The messages broadcast around the cluster.
- */
-public class SocketClusterMessage implements Serializable {
-
- private static final long serialVersionUID = 2993350408394934473L;
-
- private final String registerHost;
-
- private final boolean register;
-
- private final DataHolder dataHolder;
-
- public static SocketClusterMessage register(String registerHost, boolean register) {
- return new SocketClusterMessage(registerHost, register);
- }
-
- public static SocketClusterMessage transEvent(DataHolder transEvent) {
- return new SocketClusterMessage(transEvent);
- }
-
- public static SocketClusterMessage packet(Packet packet) {
- DataHolder d = new DataHolder(packet.getBytes());
- return new SocketClusterMessage(d);
- }
-
- /**
- * Used to construct a Child AttributeMap.
- */
- private SocketClusterMessage(String registerHost, boolean register) {
- this.registerHost = registerHost;
- this.register = register;
- this.dataHolder = null;
- }
-
- private SocketClusterMessage(DataHolder dataHolder) {
- this.dataHolder = dataHolder;
- this.registerHost = null;
- this.register = false;
- }
-
- public String toString() {
- StringBuilder sb = new StringBuilder();
- if (registerHost != null) {
- sb.append("register ");
- sb.append(register);
- sb.append(" ");
- sb.append(registerHost);
- } else {
- sb.append("transEvent ");
- }
- return sb.toString();
- }
-
- public boolean isRegisterEvent() {
- return registerHost != null;
- }
-
- public String getRegisterHost() {
- return registerHost;
- }
-
- public boolean isRegister() {
- return register;
- }
-
- public DataHolder getDataHolder() {
- return dataHolder;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java
deleted file mode 100644
index cafd5bc74..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java
+++ /dev/null
@@ -1,39 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-/**
- * The current state of this cluster member.
- */
-public class SocketClusterStatus {
-
- private final int currentGroupSize;
- private final int txnIncoming;
- private final int txtOutgoing;
-
- public SocketClusterStatus(int currentGroupSize, int txnIncoming, int txnOutgoing) {
- this.currentGroupSize = currentGroupSize;
- this.txnIncoming = txnIncoming;
- this.txtOutgoing = txnOutgoing;
- }
-
- /**
- * Return the number of members of the cluster currently online.
- */
- public int getCurrentGroupSize() {
- return currentGroupSize;
- }
-
- /**
- * Return the number of Remote transactions received.
- */
- public int getTxnIncoming() {
- return txnIncoming;
- }
-
- /**
- * Return the number of transactions sent to the cluster.
- */
- public int getTxtOutgoing() {
- return txtOutgoing;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java
deleted file mode 100644
index 81f8fd154..000000000
--- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package com.avaje.ebeaninternal.server.cluster.socket;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.ObjectInputStream;
-import java.io.OutputStream;
-import java.net.Socket;
-
-/**
- * The client side of a TCP Sockect connection.
- */
-class SocketConnection {
-
- /**
- * The underlying ObjectInputStream.
- */
- ObjectInputStream ois;
-
- /**
- * The underlying inputStream.
- */
- InputStream is;
-
- /**
- * The underlying outputStream.
- */
- OutputStream os;
-
- /**
- * The underlying socket.
- */
- Socket socket;
-
- /**
- * Create for a given Socket.
- */
- public SocketConnection(Socket socket) throws IOException {
- this.is = socket.getInputStream();
- this.os = socket.getOutputStream();
- this.socket = socket;
- }
-
- /**
- * Disconnect from the server.
- */
- public void disconnect() throws IOException {
- os.flush();
- socket.close();
- }
-
- /**
- * Flush the outputStream.
- */
- public void flush() throws IOException {
- os.flush();
- }
-
- /**
- * Read an object from the object input stream.
- */
- public Object readObject() throws IOException, ClassNotFoundException {
- return getObjectInputStream().readObject();
- }
-
- /**
- * Get the object input stream.
- */
- public ObjectInputStream getObjectInputStream() throws IOException {
- if (ois == null) {
- ois = new ObjectInputStream(is);
- }
- return ois;
- }
-
-}
diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java
index f44dfdefc..018939b9d 100644
--- a/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java
+++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java
@@ -28,17 +28,15 @@ import java.util.List;
* size of data sent around the network.
*
*/
-public class BeanPersistIds implements Serializable {
+public class BeanPersistIds {
- private static final long serialVersionUID = 8389469180931531409L;
-
- private transient BeanDescriptor beanDescriptor;
+ private final BeanDescriptor beanDescriptor;
private final String descriptorId;
- private ArrayList insertIds;
- private ArrayList updateIds;
- private ArrayList deleteIds;
+ private List