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 index 27c43a9cd..acefd636d 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java @@ -15,120 +15,118 @@ import java.net.Socket; */ class SocketClient { - private static final Logger logger = LoggerFactory.getLogger(SocketClient.class); - - private final InetSocketAddress address; - - private final String hostPort; - - private boolean online; + private static final Logger logger = LoggerFactory.getLogger(SocketClient.class); - 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(); - } + private final InetSocketAddress address; - public String toString() { - return address.toString(); - } + private final String hostPort; - public String getHostPort() { - return hostPort; - } - - public int getPort() { - return address.getPort(); - } - - public boolean isOnline() { - return online; - } + private boolean online; - public void setOnline(boolean online) throws IOException { - if (online){ - setOnline(); - } else { - disconnect(); - } - } + private Socket socket; + private OutputStream os; + private ObjectOutputStream oos; - - /** - * 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 { + /** + * Construct with an IP address and port. + */ + public SocketClient(InetSocketAddress address) { + this.address = address; + this.hostPort = address.getHostName() + ":" + address.getPort(); + } - if (online){ - writeObject(msg); - } - } - - private void writeObject(Object object) throws IOException { - if (oos == null){ - this.oos = new ObjectOutputStream(os); - } - oos.writeObject(object); - oos.flush(); - } - + 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 index 8d3945b9d..15fcf1c13 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java @@ -10,7 +10,6 @@ import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.persistence.PersistenceException; import java.io.EOFException; import java.io.IOException; import java.io.InterruptedIOException; @@ -91,13 +90,8 @@ public class SocketClusterBroadcast implements ClusterBroadcast { public void startup(ClusterManager clusterManager) { this.clusterManager = clusterManager; - try { - listener.startListening(); - register(); - - } catch (IOException e) { - throw new PersistenceException(e); - } + listener.startListening(); + register(); } public void shutdown() { @@ -184,7 +178,7 @@ public class SocketClusterBroadcast implements ClusterBroadcast { /** * Process an incoming Cluster message. */ - protected boolean process(SocketConnection request) throws IOException, ClassNotFoundException { + protected boolean process(SocketConnection request) throws ClassNotFoundException { try { SocketClusterMessage h = (SocketClusterMessage) request.readObject(); 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 index 88a99fa62..cd0824f24 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java @@ -75,7 +75,7 @@ class SocketClusterListener implements Runnable { /** * Start listening for requests. */ - public void startListening() throws IOException { + public void startListening() { logger.trace("... startListening()"); this.listenerThread.setDaemon(true); this.listenerThread.start(); 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 index f817e2040..07bfd6d11 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java @@ -1,78 +1,78 @@ package com.avaje.ebeaninternal.server.cluster.socket; -import java.io.Serializable; - 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 static final long serialVersionUID = 2993350408394934473L; - private final boolean register; - - private final DataHolder dataHolder; - - public static SocketClusterMessage register(String registerHost, boolean register){ - return new SocketClusterMessage(registerHost, register); - } + private final String registerHost; - 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; - } + private final boolean register; - 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; - } + private final DataHolder dataHolder; - public String getRegisterHost() { - return registerHost; - } + public static SocketClusterMessage register(String registerHost, boolean register) { + return new SocketClusterMessage(registerHost, register); + } - public boolean isRegister() { - return register; - } + public static SocketClusterMessage transEvent(DataHolder transEvent) { + return new SocketClusterMessage(transEvent); + } - public DataHolder getDataHolder() { - return dataHolder; + 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 index 209d528b6..cafd5bc74 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java @@ -1,41 +1,39 @@ package com.avaje.ebeaninternal.server.cluster.socket; /** - * The current state of this cluster member. - * - * @author rbygrave + * 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; - } + private final int currentGroupSize; + private final int txnIncoming; + private final int txtOutgoing; - /** - * Return the number of Remote transactions received. - */ - public int getTxnIncoming() { - return txnIncoming; - } + 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; + } - /** - * 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 index 929a11e17..81f8fd154 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java @@ -3,7 +3,6 @@ package com.avaje.ebeaninternal.server.cluster.socket; import java.io.IOException; import java.io.InputStream; import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.io.OutputStream; import java.net.Socket; @@ -12,118 +11,65 @@ import java.net.Socket; */ class SocketConnection { - /** - * The object underlying objectOutputStream. - */ - ObjectOutputStream oos; - - /** - * 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(); - } + /** + * The underlying ObjectInputStream. + */ + ObjectInputStream ois; - /** - * Read an object from the object input stream. - */ - public Object readObject() throws IOException, ClassNotFoundException { - return getObjectInputStream().readObject(); - } + /** + * The underlying inputStream. + */ + InputStream is; - /** - * Write an object to the object output stream. - */ - public ObjectOutputStream writeObject(Object object) throws IOException { - ObjectOutputStream oos = getObjectOutputStream(); - oos.writeObject(object); - return oos; - } - - /** - * Get the object output stream. - */ - public ObjectOutputStream getObjectOutputStream() throws IOException { - if (oos == null){ - oos = new ObjectOutputStream(os); - } - return oos; - } - - /** - * Get the object input stream. - */ - public ObjectInputStream getObjectInputStream() throws IOException { - if (ois == null){ - ois = new ObjectInputStream(is); - } - return ois; - } - - - /** - * Set the ObjectInputStream to use. - */ - public void setObjectInputStream(ObjectInputStream ois) { - this.ois = ois; - } + /** + * The underlying outputStream. + */ + OutputStream os; - /** - * Set the ObjectOutputStream to use. - */ - public void setObjectOutputStream(ObjectOutputStream oos) { - this.oos = oos; - } + /** + * The underlying socket. + */ + Socket socket; - /** - * Return the underlying input stream. - */ - public InputStream getInputStream() throws IOException { - return is; - } - - /** - * Return the underlying output stream. - */ - public OutputStream getOutputStream() throws IOException { - return os; + /** + * 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; + } }