- * In general terms we break up a potentially large object like - * RemoteTransactionEvent into many smaller BinaryMessages. This is so that if - * they don't all fit on a single Packet we can easily break them up and put - * them on multiple packets. - *
- *- * Also note that for the Multicast approach a Packet will generally contain - * many messages each directed to different members of the cluster. So it would - * be common for many Ack, Resend and Control messages to all be contained in a - * single packet. - *
- */ -public class BinaryMessage { - - public static final int TYPE_MSGCONTROL = 0; - public static final int TYPE_BEANIUD = 1; - public static final int TYPE_TABLEIUD = 2; - public static final int TYPE_BEANDELTA = 3; - public static final int TYPE_BEANPATHUPDATE = 4; - - public static final int TYPE_MSGACK = 8; - public static final int TYPE_MSGRESEND = 9; - - private final ByteArrayOutputStream buffer; - private final DataOutputStream os; - private byte[] bytes; - - /** - * Create with an estimated buffer size. - */ - public BinaryMessage(int bufSize) { - this.buffer = new ByteArrayOutputStream(bufSize); - this.os = new DataOutputStream(buffer); - } - - /** - * Return the DataOutputStream to write content to. - */ - public DataOutputStream getOs() { - return os; - } - - /** - * Return all the content as a byte array. - */ - public byte[] getByteArray() { - if (bytes == null) { - bytes = buffer.toByteArray(); - } - return bytes; - } -} +package com.avaje.ebeaninternal.server.cluster; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; + +/** + * Represents a relatively small independent message. + *+ * In general terms we break up a potentially large object like + * RemoteTransactionEvent into many smaller BinaryMessages. This is so that if + * they don't all fit on a single Packet we can easily break them up and put + * them on multiple packets. + *
+ *+ * Also note that for the Multicast approach a Packet will generally contain + * many messages each directed to different members of the cluster. So it would + * be common for many Ack, Resend and Control messages to all be contained in a + * single packet. + *
+ */ +public class BinaryMessage { + + public static final int TYPE_MSGCONTROL = 0; + public static final int TYPE_BEANIUD = 1; + public static final int TYPE_TABLEIUD = 2; + public static final int TYPE_BEANDELTA = 3; + public static final int TYPE_BEANPATHUPDATE = 4; + + public static final int TYPE_MSGACK = 8; + public static final int TYPE_MSGRESEND = 9; + + private final ByteArrayOutputStream buffer; + private final DataOutputStream os; + private byte[] bytes; + + /** + * Create with an estimated buffer size. + */ + public BinaryMessage(int bufSize) { + this.buffer = new ByteArrayOutputStream(bufSize); + this.os = new DataOutputStream(buffer); + } + + /** + * Return the DataOutputStream to write content to. + */ + public DataOutputStream getOs() { + return os; + } + + /** + * Return all the content as a byte array. + */ + public byte[] getByteArray() { + if (bytes == null) { + bytes = buffer.toByteArray(); + } + return bytes; + } +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/BinaryMessageList.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/BinaryMessageList.java index 3c0e36fe0..0c89a1611 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/BinaryMessageList.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/BinaryMessageList.java @@ -1,23 +1,23 @@ -package com.avaje.ebeaninternal.server.cluster; - -import java.util.ArrayList; -import java.util.List; - -/** - * Holds a List of BinaryMessage's. - * - * @author rbygrave - */ -public class BinaryMessageList { - - ArrayList- * 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 short packetType; - - /** - * The PacketId. - */ - protected long packetId; - - /** - * The timestamp the Packet was created. - */ - protected long timestamp; - - /** - * The EbeanServer name this relates to if relevant. - */ - protected 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; - } - - -} +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 short packetType; + + /** + * The PacketId. + */ + protected long packetId; + + /** + * The timestamp the Packet was created. + */ + protected long timestamp; + + /** + * The EbeanServer name this relates to if relevant. + */ + protected 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 index 9dac59a44..ab8fd5fa7 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java @@ -1,69 +1,69 @@ -package com.avaje.ebeaninternal.server.cluster; - -import java.io.DataInput; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -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; - -/** - * A Packet that contains Ack, Resend and Control messages. - * - * @author rbygrave - */ -public class PacketMessages extends Packet { - - private final ArrayList- * 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; - - case BinaryMessage.TYPE_BEANDELTA: - event.addBeanDelta(BeanDelta.readBinaryMessage(server, dataInput)); - break; - - default: - throw new RuntimeException("Invalid Transaction msgType "+msgType); - } - } - -} +package com.avaje.ebeaninternal.server.cluster; + +import java.io.DataInput; +import java.io.IOException; + +import com.avaje.ebeaninternal.api.SpiEbeanServer; +import com.avaje.ebeaninternal.api.TransactionEventTable.TableIUD; +import com.avaje.ebeaninternal.server.transaction.BeanDelta; +import com.avaje.ebeaninternal.server.transaction.BeanPersistIds; +import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; + +/** + * 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; + + case BinaryMessage.TYPE_BEANDELTA: + event.addBeanDelta(BeanDelta.readBinaryMessage(server, dataInput)); + break; + + default: + throw new RuntimeException("Invalid Transaction msgType "+msgType); + } + } + +} 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 index b35d8e6c5..8483da8a7 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java @@ -1,43 +1,43 @@ -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. - * - * @author rbygrave - */ -public class AckResendMessages { - - ArrayList- * 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. - * - * @author rbygrave - */ -public class IncomingPacketsLastAck { - - private HashMap+ * 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. + * + * @author rbygrave + */ +public class IncomingPacketsLastAck { + + private HashMap- * 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. - *
- * - * @author rbygrave - * - */ -public class IncomingPacketsProcessed { - - private final ConcurrentHashMap- * 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 ArrayList+ * 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. + *
+ * + * @author rbygrave + * + */ +public class IncomingPacketsProcessed { + + private final ConcurrentHashMap+ * 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 ArrayList- * Ideally you want to see relatively low Re-send statistics. - *
- * - * @author rbygrave - * - */ -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() { - - 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; - } - -} +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. + *
+ * + * @author rbygrave + * + */ +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() { + + 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 index f6a2fc5b6..42718418e 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java @@ -1,14 +1,14 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.io.IOException; - -import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; - -public interface Message { - - public void writeBinaryMessage(BinaryMessageList msgList) throws IOException; - - public boolean isControlMessage(); - - public String getToHostPort(); -} +package com.avaje.ebeaninternal.server.cluster.mcast; + +import java.io.IOException; + +import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; + +public interface Message { + + public void writeBinaryMessage(BinaryMessageList msgList) throws IOException; + + public boolean isControlMessage(); + + public 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 index 6266c0dd5..53cf1c93c 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java @@ -1,57 +1,57 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.io.DataInput; -import java.io.DataOutputStream; -import java.io.IOException; - -import com.avaje.ebeaninternal.server.cluster.BinaryMessage; -import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; - -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); - } -} +package com.avaje.ebeaninternal.server.cluster.mcast; + +import java.io.DataInput; +import java.io.DataOutputStream; +import java.io.IOException; + +import com.avaje.ebeaninternal.server.cluster.BinaryMessage; +import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; + +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 index d5494d3aa..2c2053ac3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java @@ -1,73 +1,73 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.io.DataInput; -import java.io.DataOutputStream; -import java.io.IOException; - -import com.avaje.ebeaninternal.server.cluster.BinaryMessage; -import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; - -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); - } -} +package com.avaje.ebeaninternal.server.cluster.mcast; + +import java.io.DataInput; +import java.io.DataOutputStream; +import java.io.IOException; + +import com.avaje.ebeaninternal.server.cluster.BinaryMessage; +import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; + +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 index 700c8a2a3..00d6e1cb3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java @@ -1,77 +1,77 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.io.DataInput; -import java.io.DataOutputStream; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -import com.avaje.ebeaninternal.server.cluster.BinaryMessage; -import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; - -public class MessageResend implements Message { - - private final String toHostPort; - - private final List- * These are held until we receive ACKs from the other members of the cluster to - * say they have received the packets. - *
- * - * @author rbygrave - * - */ -public class OutgoingPacketsCache { - - private final Map+ * These are held until we receive ACKs from the other members of the cluster to + * say they have received the packets. + *
+ * + * @author rbygrave + * + */ +public class OutgoingPacketsCache { + + private final Map- * 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); - } - } - +package com.avaje.ebeaninternal.server.cluster.socket; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +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 index 1dde45140..db5b47aa8 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 @@ -1,139 +1,139 @@ -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 boolean send(SocketClusterMessage msg) throws IOException { - - if (online){ - writeObject(msg); - return true; - - } else { - return false; - } - - } - - private void writeObject(Object object) throws IOException { - if (oos == null){ - this.oos = new ObjectOutputStream(os); - } - oos.writeObject(object); - oos.flush(); - } - - - -} +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 boolean send(SocketClusterMessage msg) throws IOException { + + if (online){ + writeObject(msg); + return true; + + } else { + return false; + } + + } + + 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 d96bb7ef0..3b73b6e3a 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 @@ -1,256 +1,256 @@ -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 javax.persistence.PersistenceException; -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- * 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() throws IOException { - 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); - } - } - } - -} +package com.avaje.ebeaninternal.server.cluster.socket; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketException; + +import com.avaje.ebeaninternal.server.lib.DaemonThreadPool; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * 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() throws IOException { + 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 index 71bc8b572..f817e2040 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; - -/** - * 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; - } - -} +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; + +/** + * 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 index fa06064db..209d528b6 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,41 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -/** - * The current state of this cluster member. - * - * @author rbygrave - */ -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; - } - -} +package com.avaje.ebeaninternal.server.cluster.socket; + +/** + * The current state of this cluster member. + * + * @author rbygrave + */ +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 index aeef8e95b..929a11e17 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 @@ -1,129 +1,129 @@ -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; - -/** - * The client side of a TCP Sockect connection. - */ -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(); - } - - /** - * Read an object from the object input stream. - */ - public Object readObject() throws IOException, ClassNotFoundException { - return getObjectInputStream().readObject(); - } - - /** - * 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; - } - - /** - * Set the ObjectOutputStream to use. - */ - public void setObjectOutputStream(ObjectOutputStream oos) { - this.oos = oos; - } - - /** - * Return the underlying input stream. - */ - public InputStream getInputStream() throws IOException { - return is; - } - - /** - * Return the underlying output stream. - */ - public OutputStream getOutputStream() throws IOException { - return os; - } - -} +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; + +/** + * The client side of a TCP Sockect connection. + */ +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(); + } + + /** + * Read an object from the object input stream. + */ + public Object readObject() throws IOException, ClassNotFoundException { + return getObjectInputStream().readObject(); + } + + /** + * 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; + } + + /** + * Set the ObjectOutputStream to use. + */ + public void setObjectOutputStream(ObjectOutputStream oos) { + this.oos = oos; + } + + /** + * Return the underlying input stream. + */ + public InputStream getInputStream() throws IOException { + return is; + } + + /** + * Return the underlying output stream. + */ + public OutputStream getOutputStream() throws IOException { + return os; + } + +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/BasicTypeConverter.java b/src/main/java/com/avaje/ebeaninternal/server/core/BasicTypeConverter.java index 7f0f6ec76..51a2daf39 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/BasicTypeConverter.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/BasicTypeConverter.java @@ -1,478 +1,478 @@ -package com.avaje.ebeaninternal.server.core; - -import java.io.Serializable; -import java.math.BigDecimal; -import java.math.BigInteger; -import java.sql.Timestamp; -import java.sql.Types; -import java.util.Calendar; -import java.util.UUID; - -import com.avaje.ebeaninternal.server.type.ScalarTypeUUIDBinary; - -/** - * Default implementation of TypeConverter. - *- * Converts objects to the required type if required. - *
- */ -public final class BasicTypeConverter implements Serializable { - - private static final long serialVersionUID = 7691463236204070311L; - - /** - * Type code for java.util.Calendar. - */ - public static final int UTIL_CALENDAR = -999998986; - - /** - * Type code for java.util.Date. - */ - public static final int UTIL_DATE = -999998988; - - /** - * Type code for java.math.BigInteger. - */ - public static final int MATH_BIGINTEGER = -999998987; - - /** - * Type code for an Enum type. - */ - public static final int ENUM = -999998989; - - private BasicTypeConverter() { - } - - /** - * Convert the Object to the required data type. - * - * @param value - * the Object value - * @param toDataType - * the dataType as per java.sql.Types. - */ - public static Object convert(Object value, int toDataType) { - - try { - switch (toDataType) { - case UTIL_DATE: { - return toUtilDate(value); - } - case UTIL_CALENDAR: { - return toCalendar(value); - } - case Types.BIGINT: { - return toLong(value); - } - case Types.INTEGER: { - return toInteger(value); - } - case Types.BIT: { - return toBoolean(value); - } - case Types.TINYINT: { - return toByte(value); - } - case Types.SMALLINT: { - return toShort(value); - } - case Types.NUMERIC: { - return toBigDecimal(value); - } - case Types.DECIMAL: { - return toBigDecimal(value); - } - case Types.REAL: { - return toFloat(value); - } - case Types.DOUBLE: { - return toDouble(value); - } - case Types.FLOAT: { - return toDouble(value); - } - case Types.BOOLEAN: { - return toBoolean(value); - } - case Types.TIMESTAMP: { - return toTimestamp(value); - } - case Types.DATE: { - return toDate(value); - } - case Types.VARCHAR: { - return toString(value); - } - case Types.CHAR: { - return toString(value); - } - case Types.OTHER: { - return value; - } - case Types.JAVA_OBJECT: { - return value; - } - case Types.BINARY: - case Types.LONGVARBINARY: - case Types.BLOB: { - return value; - } - case Types.LONGVARCHAR: - case Types.CLOB: { - return value; - } - default: { - String msg = "Unhandled data type [" + toDataType + "] converting [" + value + "]"; - throw new RuntimeException(msg); - } - } - } catch (ClassCastException e) { - String m = "ClassCastException converting to data type [" + toDataType + "] value [" + value + "]"; - throw new RuntimeException(m); - } - } - - /** - * Convert the value to a String. - */ - public static String toString(Object value) { - - if (value == null) { - return null; - } - if (value instanceof String) { - return (String) value; - } - if (value instanceof char[]) { - return String.valueOf((char[]) value); - } - - return value.toString(); - } - - - /** - * Convert the value to a Boolean with an explicit String true value. - */ - public static Boolean toBoolean(Object value, String dbTrueValue) { - - if (value == null) { - return null; - } - if (value instanceof Boolean) { - return (Boolean) value; - } - String s = value.toString(); - return s.equalsIgnoreCase(dbTrueValue); - } - - /** - * Convert the value to a Boolean. Can be a Boolean or the string values - * "true" or "false". - */ - public static Boolean toBoolean(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Boolean) { - return (Boolean) value; - } - - return Boolean.valueOf(value.toString()); - } - - /** - * Convert the value to a UUID. - */ - public static UUID toUUID(Object value) { - - if (value == null) { - return null; - } - if (value instanceof String) { - return UUID.fromString((String) value); - } - if (value instanceof byte[]) { - return ScalarTypeUUIDBinary.convertFromBytes((byte[])value); - } - return (UUID) value; - } - - /** - * convert the passed in object to a BigDecimal. It should be another - * numeric type. - */ - public static BigDecimal toBigDecimal(Object value) { - - if (value == null) { - return null; - } - if (value instanceof BigDecimal) { - return (BigDecimal) value; - } - return new BigDecimal(value.toString()); - } - - public static Float toFloat(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Float) { - return (Float) value; - } - if (value instanceof Number) { - return Float.valueOf(((Number) value).floatValue()); - } - return Float.valueOf(value.toString()); - } - - public static Short toShort(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Short) { - return (Short) value; - } - if (value instanceof Number) { - return Short.valueOf(((Number) value).shortValue()); - } - return Short.valueOf(value.toString()); - } - - public static Byte toByte(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Byte) { - return (Byte) value; - } - return Byte.valueOf(value.toString()); - } - - /** - * convert the passed in object to a Integer. It should be another numeric - * type. - */ - public static Integer toInteger(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Integer) { - return (Integer) value; - } - if (value instanceof Number) { - return Integer.valueOf(((Number) value).intValue()); - } - return Integer.valueOf(value.toString()); - } - - /** - * Convert the object to a Long. It should be another numeric type. - */ - public static Long toLong(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Long) { - return (Long) value; - } - if (value instanceof String) { - return Long.valueOf((String) value); - } - if (value instanceof Number) { - return Long.valueOf(((Number) value).longValue()); - } - if (value instanceof java.util.Date) { - return Long.valueOf(((java.util.Date) value).getTime()); - } - if (value instanceof Calendar) { - return Long.valueOf(((Calendar) value).getTime().getTime()); - } - return Long.valueOf(value.toString()); - } - - public static BigInteger toMathBigInteger(Object value) { - - if (value == null) { - return null; - } - if (value instanceof BigInteger) { - return (BigInteger) value; - } - return new BigInteger(value.toString()); - } - - /** - * Convert the object to a Double. It should be another numberic type. - */ - public static Double toDouble(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Double) { - return (Double) value; - } - if (value instanceof Number) { - return Double.valueOf(((Number) value).doubleValue()); - } - return Double.valueOf(value.toString()); - } - - /** - * convert the passed in object to a Timestamp. It is expected to be a - * java.sql.Date really. - */ - public static Timestamp toTimestamp(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Timestamp) { - return (Timestamp) value; - - } else if (value instanceof java.util.Date) { - // no nanos here... so hopefully ok - return new Timestamp(((java.util.Date) value).getTime()); - - } else if (value instanceof Calendar) { - return new Timestamp(((Calendar) value).getTime().getTime()); - - } else if (value instanceof String) { - return Timestamp.valueOf((String) value); - - } else if (value instanceof Number) { - return new Timestamp(((Number) value).longValue()); - - } else { - String msg = "Unable to convert [" + value.getClass().getName() + "] into a Timestamp."; - throw new RuntimeException(msg); - } - } - - public static java.sql.Time toTime(Object value) { - - if (value == null) { - return null; - } - if (value instanceof java.sql.Time) { - return (java.sql.Time) value; - - } else if (value instanceof String) { - return java.sql.Time.valueOf((String) value); - - } else { - String m = "Unable to convert [" + value.getClass().getName() + "] into a java.sql.Date."; - throw new RuntimeException(m); - } - } - - /** - * convert the passed in object to a java sql Date. - */ - public static java.sql.Date toDate(Object value) { - - if (value == null) { - return null; - } - if (value instanceof java.sql.Date) { - return (java.sql.Date) value; - - } else if (value instanceof java.util.Date) { - return new java.sql.Date(((java.util.Date) value).getTime()); - - } else if (value instanceof Calendar) { - return new java.sql.Date(((Calendar) value).getTime().getTime()); - - } else if (value instanceof String) { - return java.sql.Date.valueOf((String) value); - - } else if (value instanceof Number) { - return new java.sql.Date(((Number) value).longValue()); - - } else { - String m = "Unable to convert [" + value.getClass().getName() + "] into a java.sql.Date."; - throw new RuntimeException(m); - } - } - - /** - * convert the passed in object to a java sql Date. - */ - public static java.util.Date toUtilDate(Object value) { - - if (value == null) { - return null; - } - if (value instanceof java.sql.Timestamp) { - // loss of nanos precision - return new java.util.Date(((java.sql.Timestamp) value).getTime()); - } - // DEVNOTE: strictly speaking do I need to convert a java.sql.Date to - // java.util.Date? equals() is symmetrical so perhaps this is not - // really required? - if (value instanceof java.sql.Date) { - return new java.util.Date(((java.sql.Date) value).getTime()); - } - if (value instanceof java.util.Date) { - return (java.util.Date) value; - - } else if (value instanceof Calendar) { - return ((Calendar) value).getTime(); - - } else if (value instanceof String) { - return new java.util.Date(Timestamp.valueOf((String) value).getTime()); - - } else if (value instanceof Number) { - return new java.util.Date(((Number) value).longValue()); - - } else { - throw new RuntimeException("Unable to convert [" + value.getClass().getName() + "] into a java.util.Date"); - } - } - - /** - * convert the passed in object to a java sql Date. - */ - public static Calendar toCalendar(Object value) { - - if (value == null) { - return null; - } - if (value instanceof Calendar) { - return (Calendar) value; - - } else if (value instanceof java.util.Date) { - java.util.Date date = ((java.util.Date) value); - return toCalendarFromDate(date); - - } else if (value instanceof String) { - java.util.Date date = toUtilDate(value); - return toCalendarFromDate(date); - - } else if (value instanceof Number) { - long timeMillis = ((Number) value).longValue(); - java.util.Date date = new java.util.Date(timeMillis); - return toCalendarFromDate(date); - - } else { - String m = "Unable to convert [" + value.getClass().getName() + "] into a java.util.Date"; - throw new RuntimeException(m); - } - } - - private static Calendar toCalendarFromDate(java.util.Date date) { - - Calendar cal = Calendar.getInstance(); - cal.setTime(date); - - return cal; - } - -} +package com.avaje.ebeaninternal.server.core; + +import java.io.Serializable; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.sql.Timestamp; +import java.sql.Types; +import java.util.Calendar; +import java.util.UUID; + +import com.avaje.ebeaninternal.server.type.ScalarTypeUUIDBinary; + +/** + * Default implementation of TypeConverter. + *+ * Converts objects to the required type if required. + *
+ */ +public final class BasicTypeConverter implements Serializable { + + private static final long serialVersionUID = 7691463236204070311L; + + /** + * Type code for java.util.Calendar. + */ + public static final int UTIL_CALENDAR = -999998986; + + /** + * Type code for java.util.Date. + */ + public static final int UTIL_DATE = -999998988; + + /** + * Type code for java.math.BigInteger. + */ + public static final int MATH_BIGINTEGER = -999998987; + + /** + * Type code for an Enum type. + */ + public static final int ENUM = -999998989; + + private BasicTypeConverter() { + } + + /** + * Convert the Object to the required data type. + * + * @param value + * the Object value + * @param toDataType + * the dataType as per java.sql.Types. + */ + public static Object convert(Object value, int toDataType) { + + try { + switch (toDataType) { + case UTIL_DATE: { + return toUtilDate(value); + } + case UTIL_CALENDAR: { + return toCalendar(value); + } + case Types.BIGINT: { + return toLong(value); + } + case Types.INTEGER: { + return toInteger(value); + } + case Types.BIT: { + return toBoolean(value); + } + case Types.TINYINT: { + return toByte(value); + } + case Types.SMALLINT: { + return toShort(value); + } + case Types.NUMERIC: { + return toBigDecimal(value); + } + case Types.DECIMAL: { + return toBigDecimal(value); + } + case Types.REAL: { + return toFloat(value); + } + case Types.DOUBLE: { + return toDouble(value); + } + case Types.FLOAT: { + return toDouble(value); + } + case Types.BOOLEAN: { + return toBoolean(value); + } + case Types.TIMESTAMP: { + return toTimestamp(value); + } + case Types.DATE: { + return toDate(value); + } + case Types.VARCHAR: { + return toString(value); + } + case Types.CHAR: { + return toString(value); + } + case Types.OTHER: { + return value; + } + case Types.JAVA_OBJECT: { + return value; + } + case Types.BINARY: + case Types.LONGVARBINARY: + case Types.BLOB: { + return value; + } + case Types.LONGVARCHAR: + case Types.CLOB: { + return value; + } + default: { + String msg = "Unhandled data type [" + toDataType + "] converting [" + value + "]"; + throw new RuntimeException(msg); + } + } + } catch (ClassCastException e) { + String m = "ClassCastException converting to data type [" + toDataType + "] value [" + value + "]"; + throw new RuntimeException(m); + } + } + + /** + * Convert the value to a String. + */ + public static String toString(Object value) { + + if (value == null) { + return null; + } + if (value instanceof String) { + return (String) value; + } + if (value instanceof char[]) { + return String.valueOf((char[]) value); + } + + return value.toString(); + } + + + /** + * Convert the value to a Boolean with an explicit String true value. + */ + public static Boolean toBoolean(Object value, String dbTrueValue) { + + if (value == null) { + return null; + } + if (value instanceof Boolean) { + return (Boolean) value; + } + String s = value.toString(); + return s.equalsIgnoreCase(dbTrueValue); + } + + /** + * Convert the value to a Boolean. Can be a Boolean or the string values + * "true" or "false". + */ + public static Boolean toBoolean(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Boolean) { + return (Boolean) value; + } + + return Boolean.valueOf(value.toString()); + } + + /** + * Convert the value to a UUID. + */ + public static UUID toUUID(Object value) { + + if (value == null) { + return null; + } + if (value instanceof String) { + return UUID.fromString((String) value); + } + if (value instanceof byte[]) { + return ScalarTypeUUIDBinary.convertFromBytes((byte[])value); + } + return (UUID) value; + } + + /** + * convert the passed in object to a BigDecimal. It should be another + * numeric type. + */ + public static BigDecimal toBigDecimal(Object value) { + + if (value == null) { + return null; + } + if (value instanceof BigDecimal) { + return (BigDecimal) value; + } + return new BigDecimal(value.toString()); + } + + public static Float toFloat(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Float) { + return (Float) value; + } + if (value instanceof Number) { + return Float.valueOf(((Number) value).floatValue()); + } + return Float.valueOf(value.toString()); + } + + public static Short toShort(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Short) { + return (Short) value; + } + if (value instanceof Number) { + return Short.valueOf(((Number) value).shortValue()); + } + return Short.valueOf(value.toString()); + } + + public static Byte toByte(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Byte) { + return (Byte) value; + } + return Byte.valueOf(value.toString()); + } + + /** + * convert the passed in object to a Integer. It should be another numeric + * type. + */ + public static Integer toInteger(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Integer) { + return (Integer) value; + } + if (value instanceof Number) { + return Integer.valueOf(((Number) value).intValue()); + } + return Integer.valueOf(value.toString()); + } + + /** + * Convert the object to a Long. It should be another numeric type. + */ + public static Long toLong(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Long) { + return (Long) value; + } + if (value instanceof String) { + return Long.valueOf((String) value); + } + if (value instanceof Number) { + return Long.valueOf(((Number) value).longValue()); + } + if (value instanceof java.util.Date) { + return Long.valueOf(((java.util.Date) value).getTime()); + } + if (value instanceof Calendar) { + return Long.valueOf(((Calendar) value).getTime().getTime()); + } + return Long.valueOf(value.toString()); + } + + public static BigInteger toMathBigInteger(Object value) { + + if (value == null) { + return null; + } + if (value instanceof BigInteger) { + return (BigInteger) value; + } + return new BigInteger(value.toString()); + } + + /** + * Convert the object to a Double. It should be another numberic type. + */ + public static Double toDouble(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Double) { + return (Double) value; + } + if (value instanceof Number) { + return Double.valueOf(((Number) value).doubleValue()); + } + return Double.valueOf(value.toString()); + } + + /** + * convert the passed in object to a Timestamp. It is expected to be a + * java.sql.Date really. + */ + public static Timestamp toTimestamp(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Timestamp) { + return (Timestamp) value; + + } else if (value instanceof java.util.Date) { + // no nanos here... so hopefully ok + return new Timestamp(((java.util.Date) value).getTime()); + + } else if (value instanceof Calendar) { + return new Timestamp(((Calendar) value).getTime().getTime()); + + } else if (value instanceof String) { + return Timestamp.valueOf((String) value); + + } else if (value instanceof Number) { + return new Timestamp(((Number) value).longValue()); + + } else { + String msg = "Unable to convert [" + value.getClass().getName() + "] into a Timestamp."; + throw new RuntimeException(msg); + } + } + + public static java.sql.Time toTime(Object value) { + + if (value == null) { + return null; + } + if (value instanceof java.sql.Time) { + return (java.sql.Time) value; + + } else if (value instanceof String) { + return java.sql.Time.valueOf((String) value); + + } else { + String m = "Unable to convert [" + value.getClass().getName() + "] into a java.sql.Date."; + throw new RuntimeException(m); + } + } + + /** + * convert the passed in object to a java sql Date. + */ + public static java.sql.Date toDate(Object value) { + + if (value == null) { + return null; + } + if (value instanceof java.sql.Date) { + return (java.sql.Date) value; + + } else if (value instanceof java.util.Date) { + return new java.sql.Date(((java.util.Date) value).getTime()); + + } else if (value instanceof Calendar) { + return new java.sql.Date(((Calendar) value).getTime().getTime()); + + } else if (value instanceof String) { + return java.sql.Date.valueOf((String) value); + + } else if (value instanceof Number) { + return new java.sql.Date(((Number) value).longValue()); + + } else { + String m = "Unable to convert [" + value.getClass().getName() + "] into a java.sql.Date."; + throw new RuntimeException(m); + } + } + + /** + * convert the passed in object to a java sql Date. + */ + public static java.util.Date toUtilDate(Object value) { + + if (value == null) { + return null; + } + if (value instanceof java.sql.Timestamp) { + // loss of nanos precision + return new java.util.Date(((java.sql.Timestamp) value).getTime()); + } + // DEVNOTE: strictly speaking do I need to convert a java.sql.Date to + // java.util.Date? equals() is symmetrical so perhaps this is not + // really required? + if (value instanceof java.sql.Date) { + return new java.util.Date(((java.sql.Date) value).getTime()); + } + if (value instanceof java.util.Date) { + return (java.util.Date) value; + + } else if (value instanceof Calendar) { + return ((Calendar) value).getTime(); + + } else if (value instanceof String) { + return new java.util.Date(Timestamp.valueOf((String) value).getTime()); + + } else if (value instanceof Number) { + return new java.util.Date(((Number) value).longValue()); + + } else { + throw new RuntimeException("Unable to convert [" + value.getClass().getName() + "] into a java.util.Date"); + } + } + + /** + * convert the passed in object to a java sql Date. + */ + public static Calendar toCalendar(Object value) { + + if (value == null) { + return null; + } + if (value instanceof Calendar) { + return (Calendar) value; + + } else if (value instanceof java.util.Date) { + java.util.Date date = ((java.util.Date) value); + return toCalendarFromDate(date); + + } else if (value instanceof String) { + java.util.Date date = toUtilDate(value); + return toCalendarFromDate(date); + + } else if (value instanceof Number) { + long timeMillis = ((Number) value).longValue(); + java.util.Date date = new java.util.Date(timeMillis); + return toCalendarFromDate(date); + + } else { + String m = "Unable to convert [" + value.getClass().getName() + "] into a java.util.Date"; + throw new RuntimeException(m); + } + } + + private static Calendar toCalendarFromDate(java.util.Date date) { + + Calendar cal = Calendar.getInstance(); + cal.setTime(date); + + return cal; + } + +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/BeanRequest.java b/src/main/java/com/avaje/ebeaninternal/server/core/BeanRequest.java index b9aad6820..f261bdee2 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/BeanRequest.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/BeanRequest.java @@ -1,132 +1,132 @@ -package com.avaje.ebeaninternal.server.core; - -import java.sql.Connection; - -import javax.persistence.PersistenceException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.avaje.ebean.EbeanServer; -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.api.SpiTransaction; - -/** - * Base class for find and persist requests. - */ -public abstract class BeanRequest { - - private static final Logger log = LoggerFactory.getLogger(BeanRequest.class); - - /** - * The server processing the request. - */ - protected final SpiEbeanServer ebeanServer; - - protected final String serverName; - - /** - * The transaction this is part of. - */ - protected SpiTransaction transaction; - - protected boolean createdTransaction; - - public BeanRequest(SpiEbeanServer ebeanServer, SpiTransaction t) { - this.ebeanServer = ebeanServer; - this.serverName = ebeanServer.getName(); - this.transaction = t; - } - - /** - * initialise an implicit transaction if one is not currently supplied. - *- * A transaction may have been passed in or active in the thread local. If - * not then create one implicitly to handle the request. - *
- */ - public abstract void initTransIfRequired(); - - /** - * A helper method for creating an implicit transaction is it is required. - *- * A transaction may have been passed in or active in the thread local. If - * not then create one implicitly to handle the request. - *
- */ - public void createImplicitTransIfRequired(boolean readOnlyTransaction) { - if (transaction == null) { - transaction = ebeanServer.getCurrentServerTransaction(); - if (transaction == null || !transaction.isActive()) { - // create an implicit transaction to execute this query - transaction = ebeanServer.createServerTransaction(false, -1); - createdTransaction = true; - } - } - } - - /** - * Commit this transaction if it was created for this request. - */ - public void commitTransIfRequired() { - if (createdTransaction) { - transaction.commit(); - } - } - - /** - * Rollback the transaction if it was created for this request. - */ - public void rollbackTransIfRequired() { - if (createdTransaction) { - try { - transaction.rollback(); - } catch (PersistenceException e) { - // Just log this and carry on. A previous exception has been - // thrown and if this rollback throws exception it likely means - // that the connection is broken (and the datasource and db will cleanup) - log.error("Error trying to rollack a transaction (after a prior exception thrown)", e); - } - } - } - - /** - * Return the server processing the request. Made available for - * BeanController and BeanFinder. - */ - public EbeanServer getEbeanServer() { - return ebeanServer; - } - - public SpiEbeanServer getServer() { - return ebeanServer; - } - - /** - * Return the Transaction associated with this request. - */ - public SpiTransaction getTransaction() { - return transaction; - } - - /** - * Returns the connection from the Transaction. - */ - public Connection getConnection() { - return transaction.getInternalConnection(); - } - - /** - * Return true if SQL should be logged for this transaction. - */ - public boolean isLogSql() { - return transaction.isLogSql(); - } - - /** - * Return true if SUMMARY information should be logged for this transaction. - */ - public boolean isLogSummary() { - return transaction.isLogSummary(); - } -} +package com.avaje.ebeaninternal.server.core; + +import java.sql.Connection; + +import javax.persistence.PersistenceException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.avaje.ebean.EbeanServer; +import com.avaje.ebeaninternal.api.SpiEbeanServer; +import com.avaje.ebeaninternal.api.SpiTransaction; + +/** + * Base class for find and persist requests. + */ +public abstract class BeanRequest { + + private static final Logger log = LoggerFactory.getLogger(BeanRequest.class); + + /** + * The server processing the request. + */ + protected final SpiEbeanServer ebeanServer; + + protected final String serverName; + + /** + * The transaction this is part of. + */ + protected SpiTransaction transaction; + + protected boolean createdTransaction; + + public BeanRequest(SpiEbeanServer ebeanServer, SpiTransaction t) { + this.ebeanServer = ebeanServer; + this.serverName = ebeanServer.getName(); + this.transaction = t; + } + + /** + * initialise an implicit transaction if one is not currently supplied. + *+ * A transaction may have been passed in or active in the thread local. If + * not then create one implicitly to handle the request. + *
+ */ + public abstract void initTransIfRequired(); + + /** + * A helper method for creating an implicit transaction is it is required. + *+ * A transaction may have been passed in or active in the thread local. If + * not then create one implicitly to handle the request. + *
+ */ + public void createImplicitTransIfRequired(boolean readOnlyTransaction) { + if (transaction == null) { + transaction = ebeanServer.getCurrentServerTransaction(); + if (transaction == null || !transaction.isActive()) { + // create an implicit transaction to execute this query + transaction = ebeanServer.createServerTransaction(false, -1); + createdTransaction = true; + } + } + } + + /** + * Commit this transaction if it was created for this request. + */ + public void commitTransIfRequired() { + if (createdTransaction) { + transaction.commit(); + } + } + + /** + * Rollback the transaction if it was created for this request. + */ + public void rollbackTransIfRequired() { + if (createdTransaction) { + try { + transaction.rollback(); + } catch (PersistenceException e) { + // Just log this and carry on. A previous exception has been + // thrown and if this rollback throws exception it likely means + // that the connection is broken (and the datasource and db will cleanup) + log.error("Error trying to rollack a transaction (after a prior exception thrown)", e); + } + } + } + + /** + * Return the server processing the request. Made available for + * BeanController and BeanFinder. + */ + public EbeanServer getEbeanServer() { + return ebeanServer; + } + + public SpiEbeanServer getServer() { + return ebeanServer; + } + + /** + * Return the Transaction associated with this request. + */ + public SpiTransaction getTransaction() { + return transaction; + } + + /** + * Returns the connection from the Transaction. + */ + public Connection getConnection() { + return transaction.getInternalConnection(); + } + + /** + * Return true if SQL should be logged for this transaction. + */ + public boolean isLogSql() { + return transaction.isLogSql(); + } + + /** + * Return true if SUMMARY information should be logged for this transaction. + */ + public boolean isLogSummary() { + return transaction.isLogSummary(); + } +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/BootupClasses.java b/src/main/java/com/avaje/ebeaninternal/server/core/BootupClasses.java index 2a28160aa..d8c6d6cff 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/BootupClasses.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/BootupClasses.java @@ -1,392 +1,392 @@ -package com.avaje.ebeaninternal.server.core; - -import java.lang.annotation.Annotation; -import java.util.ArrayList; -import java.util.List; - -import javax.persistence.Embeddable; -import javax.persistence.Entity; -import javax.persistence.Table; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.avaje.ebean.config.CompoundType; -import com.avaje.ebean.config.ScalarTypeConverter; -import com.avaje.ebean.config.ServerConfig; -import com.avaje.ebean.event.BeanFinder; -import com.avaje.ebean.event.BeanPersistController; -import com.avaje.ebean.event.BeanPersistListener; -import com.avaje.ebean.event.BeanQueryAdapter; -import com.avaje.ebean.event.ServerConfigStartup; -import com.avaje.ebean.event.TransactionEventListener; -import com.avaje.ebeaninternal.server.type.ScalarType; -import com.avaje.ebeaninternal.server.util.ClassPathSearchMatcher; - -/** - * Interesting classes for a EbeanServer such as Embeddable, Entity, - * ScalarTypes, Finders, Listeners and Controllers. - */ -public class BootupClasses implements ClassPathSearchMatcher { - - private static final Logger logger = LoggerFactory.getLogger(BootupClasses.class); - - private ArrayList- * This includes ScalarType, BeanController, BeanFinder and BeanListener. - *
- */ - private boolean isInterestingInterface(Class cls) { - - boolean interesting = false; - - if (BeanPersistController.class.isAssignableFrom(cls)) { - beanControllerList.add(cls); - interesting = true; - } - - if (TransactionEventListener.class.isAssignableFrom(cls)) { - transactionEventListenerList.add(cls); - interesting = true; - } - - if (ScalarType.class.isAssignableFrom(cls)) { - scalarTypeList.add(cls); - interesting = true; - } - - if (ScalarTypeConverter.class.isAssignableFrom(cls)) { - scalarConverterList.add(cls); - interesting = true; - } - - if (CompoundType.class.isAssignableFrom(cls)) { - compoundTypeList.add(cls); - interesting = true; - } - - if (BeanFinder.class.isAssignableFrom(cls)) { - beanFinderList.add(cls); - interesting = true; - } - - if (BeanPersistListener.class.isAssignableFrom(cls)) { - beanListenerList.add(cls); - interesting = true; - } - - if (BeanQueryAdapter.class.isAssignableFrom(cls)) { - beanQueryAdapterList.add(cls); - interesting = true; - } - - if (ServerConfigStartup.class.isAssignableFrom(cls)){ - serverConfigStartupList.add(cls); - interesting = true; - } - - return interesting; - } - - private boolean isEntity(Class cls) { - - Annotation ann = cls.getAnnotation(Entity.class); - if (ann != null) { - return true; - } - ann = cls.getAnnotation(Table.class); - if (ann != null) { - return true; - } - return false; - } - - private boolean isEmbeddable(Class cls) { - - Annotation ann = cls.getAnnotation(Embeddable.class); - if (ann != null) { - return true; - } - return false; - } -} +package com.avaje.ebeaninternal.server.core; + +import java.lang.annotation.Annotation; +import java.util.ArrayList; +import java.util.List; + +import javax.persistence.Embeddable; +import javax.persistence.Entity; +import javax.persistence.Table; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.avaje.ebean.config.CompoundType; +import com.avaje.ebean.config.ScalarTypeConverter; +import com.avaje.ebean.config.ServerConfig; +import com.avaje.ebean.event.BeanFinder; +import com.avaje.ebean.event.BeanPersistController; +import com.avaje.ebean.event.BeanPersistListener; +import com.avaje.ebean.event.BeanQueryAdapter; +import com.avaje.ebean.event.ServerConfigStartup; +import com.avaje.ebean.event.TransactionEventListener; +import com.avaje.ebeaninternal.server.type.ScalarType; +import com.avaje.ebeaninternal.server.util.ClassPathSearchMatcher; + +/** + * Interesting classes for a EbeanServer such as Embeddable, Entity, + * ScalarTypes, Finders, Listeners and Controllers. + */ +public class BootupClasses implements ClassPathSearchMatcher { + + private static final Logger logger = LoggerFactory.getLogger(BootupClasses.class); + + private ArrayList+ * This includes ScalarType, BeanController, BeanFinder and BeanListener. + *
+ */ + private boolean isInterestingInterface(Class cls) { + + boolean interesting = false; + + if (BeanPersistController.class.isAssignableFrom(cls)) { + beanControllerList.add(cls); + interesting = true; + } + + if (TransactionEventListener.class.isAssignableFrom(cls)) { + transactionEventListenerList.add(cls); + interesting = true; + } + + if (ScalarType.class.isAssignableFrom(cls)) { + scalarTypeList.add(cls); + interesting = true; + } + + if (ScalarTypeConverter.class.isAssignableFrom(cls)) { + scalarConverterList.add(cls); + interesting = true; + } + + if (CompoundType.class.isAssignableFrom(cls)) { + compoundTypeList.add(cls); + interesting = true; + } + + if (BeanFinder.class.isAssignableFrom(cls)) { + beanFinderList.add(cls); + interesting = true; + } + + if (BeanPersistListener.class.isAssignableFrom(cls)) { + beanListenerList.add(cls); + interesting = true; + } + + if (BeanQueryAdapter.class.isAssignableFrom(cls)) { + beanQueryAdapterList.add(cls); + interesting = true; + } + + if (ServerConfigStartup.class.isAssignableFrom(cls)){ + serverConfigStartupList.add(cls); + interesting = true; + } + + return interesting; + } + + private boolean isEntity(Class cls) { + + Annotation ann = cls.getAnnotation(Entity.class); + if (ann != null) { + return true; + } + ann = cls.getAnnotation(Table.class); + if (ann != null) { + return true; + } + return false; + } + + private boolean isEmbeddable(Class cls) { + + Annotation ann = cls.getAnnotation(Embeddable.class); + if (ann != null) { + return true; + } + return false; + } +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DatabasePlatformFactory.java b/src/main/java/com/avaje/ebeaninternal/server/core/DatabasePlatformFactory.java index 8ae2b34a3..d2600ebc3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DatabasePlatformFactory.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DatabasePlatformFactory.java @@ -1,166 +1,166 @@ -package com.avaje.ebeaninternal.server.core; - -import java.sql.Connection; -import java.sql.DatabaseMetaData; -import java.sql.SQLException; - -import javax.persistence.PersistenceException; -import javax.sql.DataSource; - -import com.avaje.ebean.config.ServerConfig; -import com.avaje.ebean.config.dbplatform.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Create a DatabasePlatform from the configuration. - *- * Will used platform name or use the meta data from the JDBC driver to - * determine the platform automatically. - *
- */ -public class DatabasePlatformFactory { - - private static final Logger logger = LoggerFactory.getLogger(DatabasePlatformFactory.class); - - /** - * Create the appropriate database specific platform. - */ - public DatabasePlatform create(ServerConfig serverConfig) { - - try { - - if (serverConfig.getDatabasePlatformName() != null) { - // choose based on dbName - return byDatabaseName(serverConfig.getDatabasePlatformName()); - - } - if (serverConfig.getDataSourceConfig().isOffline()) { - String m = "You must specify a DatabasePlatformName when you are offline"; - throw new PersistenceException(m); - } - // guess using meta data from driver - return byDataSource(serverConfig.getDataSource()); - - } catch (Exception ex) { - throw new PersistenceException(ex); - } - } - - /** - * Lookup the platform by name. - */ - private DatabasePlatform byDatabaseName(String dbName) throws SQLException { - - dbName = dbName.toLowerCase(); - if (dbName.equals("postgres") || dbName.equals("postgres9")) { - return new PostgresPlatform(); - } - if (dbName.equals("postgres8") || dbName.equals("postgres83")) { - return new Postgres8Platform(); - } - if (dbName.equals("oracle9")) { - return new Oracle9Platform(); - } - if (dbName.equals("oracle") || dbName.equals("oracle10")) { - return new Oracle10Platform(); - } - if (dbName.equals("sqlserver2005")) { - return new MsSqlServer2005Platform(); - } - if (dbName.equals("sqlserver2000")) { - return new MsSqlServer2000Platform(); - } - if (dbName.equals("sqlanywhere")) { - return new SqlAnywherePlatform(); - } - if (dbName.equals("db2")) { - return new DB2Platform(); - } - if (dbName.equals("mysql")) { - return new MySqlPlatform(); - } - - if (dbName.equals("sqlite")) { - return new SQLitePlatform(); - } - - throw new RuntimeException("database platform " + dbName + " is not known?"); - } - - /** - * Use JDBC DatabaseMetaData to determine the platform. - */ - private DatabasePlatform byDataSource(DataSource dataSource) { - - Connection conn = null; - try { - conn = dataSource.getConnection(); - DatabaseMetaData metaData = conn.getMetaData(); - - return byDatabaseMeta(metaData); - - } catch (SQLException ex) { - throw new PersistenceException(ex); - - } finally { - try { - if (conn != null) { - conn.close(); - } - } catch (SQLException ex) { - logger.error(null, ex); - } - } - } - - /** - * Find the platform by the metaData.getDatabaseProductName(). - */ - private DatabasePlatform byDatabaseMeta(DatabaseMetaData metaData) throws SQLException { - - String dbProductName = metaData.getDatabaseProductName(); - dbProductName = dbProductName.toLowerCase(); - - int majorVersion = metaData.getDatabaseMajorVersion(); - - if (dbProductName.contains("oracle")) { - if (majorVersion > 9) { - return new Oracle10Platform(); - } else { - return new Oracle9Platform(); - } - } - else if (dbProductName.contains("microsoft")) { - if (majorVersion > 8) { - return new MsSqlServer2005Platform(); - } else { - return new MsSqlServer2000Platform(); - } - } - else if (dbProductName.contains("mysql")) { - return new MySqlPlatform(); - } - else if (dbProductName.contains("h2")) { - return new H2Platform(); - } - else if (dbProductName.contains("hsql database engine")) { - return new HsqldbPlatform(); - } - else if (dbProductName.contains("postgres")) { - return new PostgresPlatform(); - } - else if (dbProductName.contains("sqlite")) { - return new SQLitePlatform(); - } - else if (dbProductName.contains("db2")) { - return new DB2Platform(); - } - else if (dbProductName.contains("sql anywhere")) { - return new SqlAnywherePlatform(); - } - - // use the standard one - return new DatabasePlatform(); - } -} +package com.avaje.ebeaninternal.server.core; + +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.SQLException; + +import javax.persistence.PersistenceException; +import javax.sql.DataSource; + +import com.avaje.ebean.config.ServerConfig; +import com.avaje.ebean.config.dbplatform.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Create a DatabasePlatform from the configuration. + *+ * Will used platform name or use the meta data from the JDBC driver to + * determine the platform automatically. + *
+ */ +public class DatabasePlatformFactory { + + private static final Logger logger = LoggerFactory.getLogger(DatabasePlatformFactory.class); + + /** + * Create the appropriate database specific platform. + */ + public DatabasePlatform create(ServerConfig serverConfig) { + + try { + + if (serverConfig.getDatabasePlatformName() != null) { + // choose based on dbName + return byDatabaseName(serverConfig.getDatabasePlatformName()); + + } + if (serverConfig.getDataSourceConfig().isOffline()) { + String m = "You must specify a DatabasePlatformName when you are offline"; + throw new PersistenceException(m); + } + // guess using meta data from driver + return byDataSource(serverConfig.getDataSource()); + + } catch (Exception ex) { + throw new PersistenceException(ex); + } + } + + /** + * Lookup the platform by name. + */ + private DatabasePlatform byDatabaseName(String dbName) throws SQLException { + + dbName = dbName.toLowerCase(); + if (dbName.equals("postgres") || dbName.equals("postgres9")) { + return new PostgresPlatform(); + } + if (dbName.equals("postgres8") || dbName.equals("postgres83")) { + return new Postgres8Platform(); + } + if (dbName.equals("oracle9")) { + return new Oracle9Platform(); + } + if (dbName.equals("oracle") || dbName.equals("oracle10")) { + return new Oracle10Platform(); + } + if (dbName.equals("sqlserver2005")) { + return new MsSqlServer2005Platform(); + } + if (dbName.equals("sqlserver2000")) { + return new MsSqlServer2000Platform(); + } + if (dbName.equals("sqlanywhere")) { + return new SqlAnywherePlatform(); + } + if (dbName.equals("db2")) { + return new DB2Platform(); + } + if (dbName.equals("mysql")) { + return new MySqlPlatform(); + } + + if (dbName.equals("sqlite")) { + return new SQLitePlatform(); + } + + throw new RuntimeException("database platform " + dbName + " is not known?"); + } + + /** + * Use JDBC DatabaseMetaData to determine the platform. + */ + private DatabasePlatform byDataSource(DataSource dataSource) { + + Connection conn = null; + try { + conn = dataSource.getConnection(); + DatabaseMetaData metaData = conn.getMetaData(); + + return byDatabaseMeta(metaData); + + } catch (SQLException ex) { + throw new PersistenceException(ex); + + } finally { + try { + if (conn != null) { + conn.close(); + } + } catch (SQLException ex) { + logger.error(null, ex); + } + } + } + + /** + * Find the platform by the metaData.getDatabaseProductName(). + */ + private DatabasePlatform byDatabaseMeta(DatabaseMetaData metaData) throws SQLException { + + String dbProductName = metaData.getDatabaseProductName(); + dbProductName = dbProductName.toLowerCase(); + + int majorVersion = metaData.getDatabaseMajorVersion(); + + if (dbProductName.contains("oracle")) { + if (majorVersion > 9) { + return new Oracle10Platform(); + } else { + return new Oracle9Platform(); + } + } + else if (dbProductName.contains("microsoft")) { + if (majorVersion > 8) { + return new MsSqlServer2005Platform(); + } else { + return new MsSqlServer2000Platform(); + } + } + else if (dbProductName.contains("mysql")) { + return new MySqlPlatform(); + } + else if (dbProductName.contains("h2")) { + return new H2Platform(); + } + else if (dbProductName.contains("hsql database engine")) { + return new HsqldbPlatform(); + } + else if (dbProductName.contains("postgres")) { + return new PostgresPlatform(); + } + else if (dbProductName.contains("sqlite")) { + return new SQLitePlatform(); + } + else if (dbProductName.contains("db2")) { + return new DB2Platform(); + } + else if (dbProductName.contains("sql anywhere")) { + return new SqlAnywherePlatform(); + } + + // use the standard one + return new DatabasePlatform(); + } +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java index 39528b43b..81ba2af0b 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java @@ -1,52 +1,52 @@ -package com.avaje.ebeaninternal.server.core; - -import java.util.concurrent.TimeUnit; - -import com.avaje.ebeaninternal.api.SpiBackgroundExecutor; -import com.avaje.ebeaninternal.server.lib.DaemonScheduleThreadPool; -import com.avaje.ebeaninternal.server.lib.DaemonThreadPool; - -/** - * The default implementation of the BackgroundExecutor. - */ -public class DefaultBackgroundExecutor implements SpiBackgroundExecutor { - - private final DaemonThreadPool pool; - - private final DaemonScheduleThreadPool schedulePool; - - /** - * Construct the default implementation of BackgroundExecutor. - * - * @param corePoolSize - * the core size of the thread pool. - * @param maximumPoolSize - * the maximum pool size before jobs are queued - * @param keepAliveSecs - * the time in seconds idle threads are keep alive - * @param shutdownWaitSeconds - * the time in seconds allowed for the pool to shutdown nicely. - * After this the pool is forced to shutdown. - */ - public DefaultBackgroundExecutor(int schedulePoolSize, int corePoolSize, int maximumPoolSize, long keepAliveSecs,int shutdownWaitSeconds, String namePrefix) { - this.pool = new DaemonThreadPool(corePoolSize, maximumPoolSize, keepAliveSecs, shutdownWaitSeconds, namePrefix); - this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix+"-periodic-"); - } - - /** - * Execute a Runnable using a background thread. - */ - public void execute(Runnable r) { - pool.execute(r); - } - - public void executePeriodically(Runnable r, long delay, TimeUnit unit) { - schedulePool.scheduleWithFixedDelay(r, delay, delay, unit); - } - - public void shutdown() { - pool.shutdown(); - schedulePool.shutdown(); - } - -} +package com.avaje.ebeaninternal.server.core; + +import java.util.concurrent.TimeUnit; + +import com.avaje.ebeaninternal.api.SpiBackgroundExecutor; +import com.avaje.ebeaninternal.server.lib.DaemonScheduleThreadPool; +import com.avaje.ebeaninternal.server.lib.DaemonThreadPool; + +/** + * The default implementation of the BackgroundExecutor. + */ +public class DefaultBackgroundExecutor implements SpiBackgroundExecutor { + + private final DaemonThreadPool pool; + + private final DaemonScheduleThreadPool schedulePool; + + /** + * Construct the default implementation of BackgroundExecutor. + * + * @param corePoolSize + * the core size of the thread pool. + * @param maximumPoolSize + * the maximum pool size before jobs are queued + * @param keepAliveSecs + * the time in seconds idle threads are keep alive + * @param shutdownWaitSeconds + * the time in seconds allowed for the pool to shutdown nicely. + * After this the pool is forced to shutdown. + */ + public DefaultBackgroundExecutor(int schedulePoolSize, int corePoolSize, int maximumPoolSize, long keepAliveSecs,int shutdownWaitSeconds, String namePrefix) { + this.pool = new DaemonThreadPool(corePoolSize, maximumPoolSize, keepAliveSecs, shutdownWaitSeconds, namePrefix); + this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix+"-periodic-"); + } + + /** + * Execute a Runnable using a background thread. + */ + public void execute(Runnable r) { + pool.execute(r); + } + + public void executePeriodically(Runnable r, long delay, TimeUnit unit) { + schedulePool.scheduleWithFixedDelay(r, delay, delay, unit); + } + + public void shutdown() { + pool.shutdown(); + schedulePool.shutdown(); + } + +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBeanLoader.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBeanLoader.java index 8e2cfc385..4f9231540 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBeanLoader.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBeanLoader.java @@ -1,442 +1,442 @@ -package com.avaje.ebeaninternal.server.core; - -import java.util.ArrayList; -import java.util.List; - -import javax.persistence.EntityNotFoundException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.avaje.ebean.ExpressionList; -import com.avaje.ebean.Transaction; -import com.avaje.ebean.bean.BeanCollection; -import com.avaje.ebean.bean.EntityBean; -import com.avaje.ebean.bean.EntityBeanIntercept; -import com.avaje.ebean.bean.ObjectGraphNode; -import com.avaje.ebean.bean.PersistenceContext; -import com.avaje.ebeaninternal.api.LoadBeanBuffer; -import com.avaje.ebeaninternal.api.LoadBeanRequest; -import com.avaje.ebeaninternal.api.LoadManyRequest; -import com.avaje.ebeaninternal.api.LoadManyBuffer; -import com.avaje.ebeaninternal.api.SpiQuery; -import com.avaje.ebeaninternal.api.SpiQuery.Mode; -import com.avaje.ebeaninternal.server.deploy.BeanDescriptor; -import com.avaje.ebeaninternal.server.deploy.BeanPropertyAssocMany; -import com.avaje.ebeaninternal.server.deploy.BeanDescriptor.EntityType; -import com.avaje.ebeaninternal.server.lib.util.StringHelper; -import com.avaje.ebeaninternal.server.transaction.DefaultPersistenceContext; - -/** - * Helper to handle lazy loading and refreshing of beans. - */ -public class DefaultBeanLoader { - - private static final Logger logger = LoggerFactory.getLogger(DefaultBeanLoader.class); - - private final DefaultServer server; - - protected DefaultBeanLoader(DefaultServer server) { - this.server = server; - } - - /** - * Return a batch size that might be less than the requestedBatchSize. - *- * This means we can have large and variable requestedBatchSizes. - *
- *- * We want to restrict the number of different batch sizes as we want to - * re-use the query plan cache and get DB statement re-use. - *
- */ - private int getBatchSize(int batchSize) { - - if (batchSize == 1) { - // there is only one bean/collection to load - return 1; - } - if (batchSize <= 5) { - // anything less than 5 becomes 5 - return 5; - } - if (batchSize <= 10) { - return 10; - } - if (batchSize <= 20) { - return 20; - } - if (batchSize <= 50) { - return 50; - } - if (batchSize <= 100) { - return 100; - } - return batchSize; - } - - public void refreshMany(EntityBean parentBean, String propertyName) { - refreshMany(parentBean, propertyName, null); - } - - public void loadMany(LoadManyRequest loadRequest) { - - List