User: hiram
Date: 01/03/01 23:04:55
Added: src/main/org/jbossmq/cluster/transport/udp Datagram.java
UDPAdminStream.java UDPNodeId.java UDPStream.java
UDPTransport.java
Log:
- Changed Source references of GPL to LGPL.
- The cluster work that I had been doing is now accesible via a Pub-Sub JMS api.
- Take a look at the Cluster* examples in the sample directory to test it out.
Revision Changes Path
1.1 jbossmq/src/main/org/jbossmq/cluster/transport/udp/Datagram.java
Index: Datagram.java
===================================================================
/*
* JBossMQ, the OpenSource JMS implementation
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jbossmq.cluster.transport.udp;
import org.jbossmq.cluster.transport.SerializerUtil;
/**
* This class is used to access/build the datagram packets
* sent by the UDP transport. Since this later has to be turnned
* into a byte[] to send over UDP, this class stores it's data
* in a byte[] once it is constructed. This speeds up serialization
* since it is ready to be serialized after construction.
*
* @author Hiram Chirino ([EMAIL PROTECTED])
*
* @version $Revision: 1.1 $
*/
class Datagram {
// Field offsets in the header of the message
final static int ID_OFFSET = 0; // int
final static int LENGTH_OFFSET = 4; // int
final static int FRAGMENT_ID_OFFSET = 8; // short
final static int SENDER_ID_OFFSET = 10; // 4 bytes
final static int TOPIC_ID_OFFSET = 14; // short
final static byte MESSAGE_FLAGS_OFFSET = 16; // byte
final static int HEADER_SIZE = 17;
// MESSAGE FLAGS
public final static byte DROPPABLE_FLAG = 1;
public final static byte KEEP_ORDER_FLAG = 2;
public final static byte ADMIN_FLAG = 4;
// the serialized message
byte data[];
// Who sent this message
UDPNodeId senderId;
// nextFragment can be used to form a linked list of Datagrams
Datagram nextFragment;
/**
* This is used to build a new datagram
*/
Datagram(
int length,
short fragmentId,
UDPNodeId senderId,
short topic,
byte messageFlags,
byte payload[],
int offset,
int len) {
data = new byte[len + HEADER_SIZE];
SerializerUtil.writeIntTo(length, data, LENGTH_OFFSET);
SerializerUtil.writeShortTo(fragmentId, data, FRAGMENT_ID_OFFSET);
senderId.writeToByteArray(data, SENDER_ID_OFFSET);
SerializerUtil.writeShortTo(topic, data, TOPIC_ID_OFFSET);
data[MESSAGE_FLAGS_OFFSET] = messageFlags;
for (int i = HEADER_SIZE, j = offset; i < data.length; i++, j++)
data[i] = payload[j];
}
/**
* This is used to reconstruct a datagram that was serialized
*/
Datagram(byte datagram[], int length) {
data = new byte[length];
for (int i = 0; i < length; i++) {
data[i] = datagram[i];
}
}
/**
* Gets the original payload of the datagram.
* @return byte[]
*/
byte[] getData() {
byte[] newData = new byte[data.length - HEADER_SIZE];
for (int i = 0, j = HEADER_SIZE; j < data.length; i++, j++)
newData[i] = data[j];
return newData;
}
/**
* Calculates how many datagrams are needs to build the message.
* This is dependent on the maxFragment size that is used.
*
* @return short
*/
int getFragmentCount(int maxFragmentSize) {
return getLength() / maxFragmentSize + (getLength() % maxFragmentSize
== 0 ? 0 : 1);
}
short getFragmentId() {
return SerializerUtil.readShortFrom(data, FRAGMENT_ID_OFFSET);
}
/**
* The id of the message. All the fragments message will have the
* same Id, just different fragment ids.
*
* @return int
*/
int getId() {
return SerializerUtil.readIntFrom(data, ID_OFFSET);
}
/**
* The total length of the message.
* @return int
*/
int getLength() {
return SerializerUtil.readIntFrom(data, LENGTH_OFFSET);
}
/**
* The topic under which this message was sent.
* @return short
*/
short getTopicId() {
return SerializerUtil.readShortFrom(data, TOPIC_ID_OFFSET);
}
/**
* The id of the sender of this message.
* @return int
*/
UDPNodeId getSenderId() {
if (senderId != null)
return senderId;
UDPNodeId t = new UDPNodeId();
t.readFromByteArray(data, SENDER_ID_OFFSET);
senderId = t;
return t;
}
/**
* returns true if all the given flags are set in the message.
*/
boolean isMessageFlagsSet(int flags) {
return (data[MESSAGE_FLAGS_OFFSET] & flags) == flags;
}
/**
* Gets all the message flags.
*/
byte getMessageFlags() {
return data[MESSAGE_FLAGS_OFFSET];
}
/**
* The id of the message. All the fragments message will have the
* same Id, just different fragment ids.
*
* @return int
*/
void setId(int id) {
SerializerUtil.writeIntTo(id, data, ID_OFFSET);
}
}
1.1
jbossmq/src/main/org/jbossmq/cluster/transport/udp/UDPAdminStream.java
Index: UDPAdminStream.java
===================================================================
/*
* JBossMQ, the OpenSource JMS implementation
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jbossmq.cluster.transport.udp;
import java.net.DatagramSocket;
import java.net.DatagramPacket;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.HashMap;
import java.util.Vector;
import org.jbossmq.Log;
/**
* The UDPAdminStream class blocks on a UDP socket
* waiting for admin packets.
*
* Admin packet must fit into a single fragment. No
* packet recovery is needed.
*
* @author Hiram Chirino ([EMAIL PROTECTED])
*
* @version $Revision: 1.1 $
*/
class UDPAdminStream implements Runnable {
// The channel manager owning this Inbound channel
UDPTransport transport;
// The thread that is reciving messages.
private Thread runningThread;
// Used to stop the thread
private boolean done = false;
// We incrementaly number the datagrams we send
private int nextDatagramId = 0;
// The socket we will be listening on
DatagramSocket socket;
/**
* Starts the input thread
*/
synchronized void start() {
if (runningThread != null)
return;
done = false;
runningThread = new Thread(this, "UDPAdminStream");
runningThread.start();
}
/**
* Stops the input thread
*/
synchronized void stop() throws InterruptedException {
if (runningThread == null)
return;
done = true;
runningThread.interrupt();
runningThread.join();
runningThread = null;
}
/**
* The thread of the InboundStream
*/
public void run() {
try {
byte buffer[] = new byte[transport.maxFragmentSize +
Datagram.HEADER_SIZE];
DatagramPacket packet = new java.net.DatagramPacket(buffer,
buffer.length);
// We iterate at least every 2 seconds so that we can
// stop the thread
socket.setSoTimeout(2000);
while (!done) {
try {
// Read in a message from the network
packet.setData(buffer);
packet.setLength(buffer.length);
socket.receive(packet);
// Is the packet ok?
if (packet.getLength() < Datagram.HEADER_SIZE)
{
Log.notice("[" + this +"] Packet was
too small, dropping.");
continue;
}
// Build the Datagram object form the network
data
Datagram dg = new Datagram(packet.getData(),
packet.getLength());
dg.getSenderId().address = packet.getAddress();
// Drop the packet if this node sent this
packet
if (transport.nodeId.equals(dg.getSenderId()))
{
continue;
}
// Used to test the packet retransmision
faclities
// of the UDPTransport class. We drop 1 in 10
packets.
/*
java.util.Random r = new java.util.Random();
if (r.nextInt(100) < 10) {
Log.error("["+this+"] TEST DROP: " +
dg.getId());
continue;
}
*/
// Continue processing the datagram
fragmentArrived(dg);
} catch (java.io.InterruptedIOException e) {
} catch (InterruptedException e) {
}
}
} catch (java.io.IOException e) {
e.printStackTrace();
}
}
/**
* This places the datagram in the proper position in the
* message "Stream"
*/
synchronized private void fragmentArrived(Datagram dg) throws
InterruptedException {
transport.datagramArrived(dg);
}
/**
* Returns the next message id in the output stream.
*/
protected int getNextDatagramId() {
// I think ++ is atomic operation in Java.. (No synch needed)
return nextDatagramId++;
}
public String toString() {
return "org.jbossmq.cluster.udp.UDPAdminStream";
}
}
1.1
jbossmq/src/main/org/jbossmq/cluster/transport/udp/UDPNodeId.java
Index: UDPNodeId.java
===================================================================
/*
* JBossMQ, the OpenSource JMS implementation
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jbossmq.cluster.transport.udp;
import java.net.InetAddress;
import java.io.Serializable;
import org.jbossmq.cluster.transport.NodeId;
import org.jbossmq.cluster.transport.SerializerUtil;
/**
* This class contains all the information needed to
* identify a node in the cluster at the transport level.
* Every node in the cluster open two new UDP sockets to
* receive point to point and admin messages. That port numbers
* combined with the IP address of the machine provides
* us a the identity of the node.
*/
public class UDPNodeId implements NodeId, Serializable {
InetAddress address;
int port;
int adminPort;
/**
* UDPNodeId constructor comment.
*/
public UDPNodeId() {
super();
}
/**
* You can compare two UDPNodeIds
*/
public boolean equals(Object obj) {
try {
UDPNodeId o = (UDPNodeId)obj;
return o.address.equals(address) &&
o.port == port &&
o.adminPort == adminPort;
} catch (Throwable e) {
return false;
}
}
/**
* The Ip address hash should get good enough in most cases.
* (Bad when you are running multiple nodes on a single machine)
*/
public int hashCode() {
return address.hashCode();
}
/**
* This is used by the Datagram class.
* The address of the UDPNodeId will be set from
* the sender field of the IP packet the datagram
* was received from.
*/
public void readFromByteArray(byte data[], int pos) {
try {
port = SerializerUtil.readUShortFrom(data, pos);
adminPort = SerializerUtil.readUShortFrom(data, pos+2);
} catch ( Exception ignore ) {
}
}
/**
* This is used by the Datagram class.
* The address of the UDPNodeId does not need to be
* serialized as the IP packet allready has a source
* IP packet field that we will take advantage of.
*/
public void writeToByteArray(byte data[], int pos) {
SerializerUtil.writeUShortTo(port,data,pos);
SerializerUtil.writeUShortTo(adminPort,data,pos+2);
}
public String toString() {
return
"UDPNodeId:["+address.getHostAddress()+":"+port+":"+adminPort+"]";
}
}
1.1
jbossmq/src/main/org/jbossmq/cluster/transport/udp/UDPStream.java
Index: UDPStream.java
===================================================================
/*
* JBossMQ, the OpenSource JMS implementation
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jbossmq.cluster.transport.udp;
import java.net.DatagramSocket;
import java.net.DatagramPacket;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.HashMap;
import java.util.Vector;
import org.jbossmq.Log;
/**
* The UDPStream class blocks on a UDP socket
* waiting for input. Since a message is made up
* of multiple fragments and UDP does not garantee message
* delivery, we keep track of the "Stream" of message
* being sent by all the nodes in the cluster.
*
* The "Stream" is a linked list of MessageState objects.
* The head of the list is the oldest message that has not been completed
* yet. The tail of the list is the most recently received message from
* the node. Messages between the tail and head end could be completely
* received or in progess. Message marked as "not required to be in order"
* are sent to the cluster to be processed by the listners as soon as they are
* completed, and removed form the "Stream". When the head message is completed
* it is sent up to the cluster to be processed and it is removed from the "stream"
* When the head message is != to the tail message, this means that a datagram
* was dropped and the head message needs to be resent, so a resend request is
issued.
* When a new message arrives it is placed at the end of the message stream but
* filler message will be added to the stream to fill the space left by any dropped
* messages.
*
* @author Hiram Chirino ([EMAIL PROTECTED])
*
* @version $Revision: 1.1 $
*/
class UDPStream implements Runnable {
// The maximum size the datagram input stream can grow to.
private static final int MAX_DATAGRAM_STREAM_SIZE = 100;
// The channel manager owning this Inbound channel
UDPTransport transport;
// The thread that is reciving messages.
private Thread runningThread;
// Used to stop the thread
private boolean done = false;
// Up to how many sent packets do we cache?
private int maxSentCacheSize = 200;
// The name this stream. Used to ease debugging.
public java.lang.String name;
// We incrementaly number the datagrams we send
private int nextDatagramId = 0;
// Used to for error recovery.
private LinkedList sentCache = new LinkedList();
// The socket we will be listening on
DatagramSocket socket;
// Maps UDPNodeIds to NodeStates
private HashMap nodes = new HashMap(10);
// Keeps track of the progress of message reception
private static class MessageState {
MessageState(int id) {
this.id = id;
}
int id;
boolean arrived = false;
boolean isDropable = false;
long lastResendRequest = 0;
Datagram[] fragments;
}
// Keeps track of the state of a remote nodes in the cluster
private static class NodeState {
UDPNodeId nodeId;
LinkedList datagramStream = new LinkedList();
int lastDatagramId = 0;
boolean pastFirstPacket = false;
}
/**
* Gets the MessageState object for the given Node/Message Id
* This will create new MessageState objects as needed.
*/
private MessageState getMessageState(NodeState nodeState, int id) {
boolean rollOver = id < -5000 && nodeState.lastDatagramId > 5000;
if (!nodeState.pastFirstPacket) {
MessageState t = new MessageState(id);
nodeState.datagramStream.addLast(t);
nodeState.lastDatagramId = id;
nodeState.pastFirstPacket = true;
return t;
} else if (nodeState.lastDatagramId == id) {
// This is weird. A node might have retransmitted a packet
for a
// message that was succesfully recived the first time
(therefore
// we are not pending messages for it in the datagramStream)
if( nodeState.datagramStream.size()==0 )
return null;
// The packet id is the same as the last packet.
return (MessageState) nodeState.datagramStream.getLast();
} else if (nodeState.lastDatagramId < id || rollOver) {
// The packet id is newer than the last packet
int i = nodeState.lastDatagramId;
while (i != id) {
i++;
// Packets got dropped but we record them in the
stream anyways.
MessageState t = new MessageState(i);
nodeState.datagramStream.addLast(t);
if (nodeState.datagramStream.size() >
MAX_DATAGRAM_STREAM_SIZE) {
handleNodeUnresponsive(nodeState);
nodeState.datagramStream.removeFirst();
}
}
nodeState.lastDatagramId = id;
return (MessageState) nodeState.datagramStream.getLast();
} else {
// The packet id older than the last packet. We have to
search for it
// int the datagram stream.
Iterator i = nodeState.datagramStream.iterator();
while (i.hasNext()) {
MessageState ds = (MessageState) i.next();
if (ds.id == id)
return ds;
}
// This is weird. A node might have retransmitted a packet
for a
// message that was succesfully recived the first time
(therefore
// we are not pending messages for it in the datagramStream)
return null;
}
}
// We should do something else to inform the cluster that
// a node is being unresponsive.
// This happens when the message stream gets too long.
// TODO: make it also happen when a message sits too long on
// the stream.
private void handleNodeUnresponsive(NodeState nodeState) {
Log.error("NODE IS UNRESPONSIVE!!!");
}
/**
* Starts the input thread
*/
synchronized void start() {
if (runningThread != null)
return;
done = false;
runningThread = new Thread(this, "UDPStream-"+name);
runningThread.start();
}
/**
* Stops the input thread
*/
synchronized void stop() throws InterruptedException {
if (runningThread == null)
return;
done = true;
runningThread.interrupt();
runningThread.join();
runningThread = null;
}
/**
* The thread of the InboundStream
*/
public void run() {
try {
byte buffer[] = new byte[transport.maxFragmentSize +
Datagram.HEADER_SIZE];
DatagramPacket packet = new java.net.DatagramPacket(buffer,
buffer.length);
// We iterate at least every 500 miliseconds so that we can
// dispatch resend request messages frequently
socket.setSoTimeout(500);
while (!done) {
try {
// We might be doing this too often..
//synchronized( this ) {
Iterator i = nodes.values().iterator();
while( i.hasNext() ) {
NodeState nodeState =
(NodeState)i.next();
dispatchDatagrams(nodeState);
}
//}
// Read in a message from the network
packet.setData(buffer);
packet.setLength(buffer.length);
socket.receive(packet);
// Is the packet ok?
if (packet.getLength() < Datagram.HEADER_SIZE)
{
Log.notice("["+this+"] Packet was too
small, dropping.");
continue;
}
// Build the Datagram object form the network
data
Datagram dg = new Datagram(packet.getData(),
packet.getLength());
dg.getSenderId().address = packet.getAddress();
// Drop the packet if this node sent this
packet
if (transport.nodeId.equals(dg.getSenderId()))
{
continue;
}
// Used to test the packet retransmision
faclities
// of the UDPTransport class. We drop 1 in 10
packets.
/*
java.util.Random r = new java.util.Random();
if (r.nextInt(100) < 10) {
Log.error("["+this+"] TEST DROP: " +
dg.getId());
continue;
}
*/
// Continue processing the datagram
fragmentArrived(dg);
} catch (java.io.InterruptedIOException e) {
} catch (InterruptedException e) {
}
}
} catch (java.io.IOException e) {
e.printStackTrace();
}
}
/**
* Adds a packet to the sent packet cache
* Removes the oldest packet if the cache is full.
*/
void addToSentCache(Datagram dg) {
if( dg.isMessageFlagsSet(Datagram.DROPPABLE_FLAG) )
return;
synchronized (sentCache) {
sentCache.addFirst(dg);
if (sentCache.size() > maxSentCacheSize)
sentCache.removeLast();
}
}
/**
* This method does the bulk of the "Stream" managment.
* - it delivers messages that are complete.
* - drops messages that are incomplete and droppable.
* - Requests retransmision of the incomplete packet at the
* head of the stream.
*/
private void dispatchDatagrams(NodeState nodeState) throws
InterruptedException {
if (nodeState.datagramStream.size() == 0)
return;
boolean atFront = true;
Object lastDS = nodeState.datagramStream.getLast();
Iterator i = nodeState.datagramStream.iterator();
MessageState firstMissing = null;
Vector missingList = new Vector();
while (i.hasNext()) {
MessageState ds = (MessageState) i.next();
if (ds.arrived) {
if (atFront ||
!(ds.fragments[0].isMessageFlagsSet(Datagram.KEEP_ORDER_FLAG)) ) {
transport.datagramArrived(ds.fragments[0]);
i.remove();
}
continue;
} else {
atFront = false;
if (ds == lastDS) {
continue;
} else if (ds.isDropable) {
i.remove();
} else {
if (firstMissing == null) {
firstMissing = ds;
}
missingList.addElement(new Integer(ds.id));
}
}
}
if ( firstMissing != null && System.currentTimeMillis() -
firstMissing.lastResendRequest > 200 ) {
transport.requestResend(nodeState.nodeId,
missingList,this==transport.broadcastStream);
firstMissing.lastResendRequest = System.currentTimeMillis();
}
}
/**
* This places the datagram in the proper position in the
* message "Stream"
*/
synchronized private void fragmentArrived(Datagram dg) throws
InterruptedException {
UDPNodeId senderId = dg.getSenderId();
int id = dg.getId();
short fragmentId = dg.getFragmentId();
Log.log("["+this+"] Processing datagram: " + id);
NodeState nodeState = getNodeState(senderId);
MessageState ds = getMessageState(nodeState, id);
// We might have allready received this message.
if (ds == null || ds.arrived)
return;
if (dg.getFragmentCount(transport.maxFragmentSize) == 1) {
// The entire message was contained within this datagram
ds.arrived = true;
ds.fragments = new Datagram[1];
ds.fragments[0] = dg;
} else {
// The message is fragmented.
if (ds.fragments == null) {
ds.fragments = new
Datagram[dg.getFragmentCount(transport.maxFragmentSize)];
ds.isDropable =
dg.isMessageFlagsSet(Datagram.DROPPABLE_FLAG);
}
ds.fragments[fragmentId] = dg;
// Check to see if we have recived all the fragments?
boolean messageLoaded = true;
// iterating backwards will break us out of the loop
// quicker in the common case
for (int i = ds.fragments.length - 1; i >= 0; i--) {
if (ds.fragments[i] == null) {
messageLoaded = false;
break;
}
}
if (messageLoaded) {
// chain the fragments together
for (int i = 0; i < ds.fragments.length - 1; i++)
ds.fragments[i].nextFragment = ds.fragments[i
+ 1];
ds.arrived = true;
}
}
}
/**
* Looks for a packet in the sent cache.
* returns null if the packet could not be found.
*/
public Datagram getFromSentCache(int messageId) {
LinkedList t;
synchronized (sentCache) {
t = (LinkedList) sentCache.clone();
}
java.util.Iterator i = t.iterator();
while (i.hasNext()) {
Datagram dg = (Datagram) i.next();
if (dg.getId() == messageId) {
return dg;
}
}
return null;
}
/**
* Returns the next message id in the output stream.
*/
protected int getNextDatagramId() {
// I think ++ is atomic operation in Java.. (No synch needed)
return nextDatagramId++;
}
/**
* Gets the NodeState for the given node id.
* if none exists, a new NodeState object is created.
*/
private NodeState getNodeState(UDPNodeId nodeId) {
NodeState rc;
rc = (NodeState)nodes.get( nodeId );
if (rc==null) {
rc = new NodeState();
rc.nodeId = nodeId;
nodes.put(nodeId, rc);
}
return (NodeState)rc;
}
/**
* This is used by the AdminPacketHandler. Called when
* a packet cannot be retransmited by the host (It was droppable or
* packet was not in the packet cache).
*
* We mark the packet as droppable so it gets removed from the stream
* the next time the messages are dispatch on the stream.
*/
synchronized void removeFromDatagramStream(UDPNodeId nodeId, int datagramId) {
NodeState nodeState = getNodeState(nodeId);
Iterator i = nodeState.datagramStream.iterator();
while (i.hasNext()) {
MessageState ds = (MessageState) i.next();
if (ds.id == datagramId)
ds.isDropable = true;
}
}
/**
* This is a descriptive name for the stream. Used of debuging messages
*/
public void setName(java.lang.String newName) {
name = newName;
}
public String toString() {
return "org.jbossmq.cluster.udp.UDPStream:"+name;
}
}
1.1
jbossmq/src/main/org/jbossmq/cluster/transport/udp/UDPTransport.java
Index: UDPTransport.java
===================================================================
/*
* JBossMQ, the OpenSource JMS implementation
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jbossmq.cluster.transport.udp;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.DatagramPacket;
import java.util.Vector;
import java.util.Properties;
import java.util.LinkedList;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.ByteArrayInputStream;
import java.io.Externalizable;
import java.io.ObjectInputStream;
import org.jbossmq.Log;
import org.jbossmq.cluster.transport.InvalidConfigurationException;
import org.jbossmq.cluster.transport.Transport;
import org.jbossmq.cluster.transport.NodeId;
import org.jbossmq.cluster.transport.TransportListener;
import org.jbossmq.cluster.transport.InvalidStateException;
/**
* This Transport implements a Datagram based cluster transport.
* It Can use UDP or Multicast.
*
* As a Transport, it will fragment outbound messages that are
* too large and recobine inbound message fragments.
*
* Since Datagrams have a chance of not making it, this class
* also provides retransmision facilities.
*
* It uses two UDPStreams. The first stream is the broadcast stream
* and it receives multicast packets sent to the cluster.
* The second is a point stream and is used to directly send a message
* to this node.
*
* @author Hiram Chirino ([EMAIL PROTECTED])
*
* @version $Revision: 1.1 $
*/
public class UDPTransport implements Transport {
// The interface into the cluster we report message arivals to
private TransportListener transportListener;
// Receives the multicast messages
UDPStream broadcastStream = new UDPStream();
// Receives the messages addressed to this node
UDPStream pointStream = new UDPStream();
// Receives admin messages addressed to this node
UDPAdminStream adminStream = new UDPAdminStream();
// The broadcast address of the cluster
UDPNodeId clusterId = new UDPNodeId();
// Our unique address in the cluster
UDPNodeId nodeId = new UDPNodeId();
// Is the transport started?
boolean started = false;
// timmer Used to slow down the transmision of messages.
volatile long stopSendsTill = 0;
// The maximum size a datagram fragment can be.
int maxFragmentSize = 1024;
/**
* I like simple constructors.
*/
public UDPTransport() {
broadcastStream.transport = this;
broadcastStream.setName("Broadcast");
pointStream.transport = this;
pointStream.setName("Point");
adminStream.transport = this;
}
// Starts the transport
synchronized public void start() throws InvalidStateException {
if (broadcastStream.socket == null)
throw new InvalidStateException("The transport properties have
not been set yet.");
started = true;
broadcastStream.start();
pointStream.start();
adminStream.start();
}
// Stops the transport
synchronized public void stop() throws InterruptedException {
broadcastStream.stop();
pointStream.stop();
adminStream.stop();
started = false;
}
// Builds a datagram fragment chain with the given data
// and then sends it over the network
public void send(short channelId, byte data[], boolean droppable, boolean
keepOrder)
throws IOException, InterruptedException {
send(null, channelId, data, droppable, keepOrder);
}
//
// This is the message sent to request a packet be resent
//
static class ResendAdminDatagram implements java.io.Serializable {
ResendAdminDatagram(int j[], boolean b) {
datagramIds = j;
broadcast = b;
}
int datagramIds[];
boolean broadcast;
}
//
// This is the message sent when the requested packet is not in the cache.
//
static class ResendErrorAdminDatagram implements java.io.Serializable {
ResendErrorAdminDatagram(int j, boolean b) {
datagramId = j;
broadcast = b;
}
int datagramId;
boolean broadcast;
}
// This send is used to send the administrative messages.
public void adminSend(UDPNodeId dest, Object message) throws IOException,
InterruptedException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream os = new ObjectOutputStream(bos);
os.writeObject(message);
os.close();
byte data[] = bos.toByteArray();
Datagram dg =
new Datagram(
data.length,
(short) 0,
nodeId,
(short) 0,
(byte) (Datagram.ADMIN_FLAG | Datagram.DROPPABLE_FLAG),
data,
0,
data.length);
dg.setId(adminStream.getNextDatagramId());
Log.log("[" + this +"] Sending admin datagram: " + dg.getId());
DatagramPacket packet = new java.net.DatagramPacket(dg.data,
dg.data.length, dest.address, dest.adminPort);
adminStream.socket.send(packet);
}
void datagramArrived(Datagram dg) throws InterruptedException {
Log.log("[" + this +"] A datagram arrived");
if (dg.isMessageFlagsSet(Datagram.ADMIN_FLAG)) {
try {
Object o;
ByteArrayInputStream bais = new
ByteArrayInputStream(dg.getData());
ObjectInputStream is = new ObjectInputStream(bais);
o = is.readObject();
is.close();
if (o instanceof ResendAdminDatagram) {
ResendAdminDatagram radg =
(ResendAdminDatagram) o;
for (int i = 0; i < radg.datagramIds.length;
i++) {
Log.log("[" + this +"] RESEND REQUEST
ARRIVED: " + radg.datagramIds[i]);
resendDatagram(dg.getSenderId(),
radg.datagramIds[i], radg.broadcast);
}
} else if (o instanceof ResendErrorAdminDatagram) {
ResendErrorAdminDatagram readg =
(ResendErrorAdminDatagram) o;
Log.log("[" + this +"] RESEND REQUEST FAILED:
" + readg.datagramId);
if (readg.broadcast) {
broadcastStream.removeFromDatagramStream(dg.getSenderId(), readg.datagramId);
} else {
pointStream.removeFromDatagramStream(dg.getSenderId(), readg.datagramId);
}
}
} catch (ClassNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
} else {
// A complete datagram has arrived to us from the InputStream
// We build a object from it and send it to the
TransportListener
if (!transportListener.isListeningOn(dg.getTopicId()))
return;
byte data[];
if (dg.getFragmentCount(maxFragmentSize) == 1) {
data = dg.getData();
} else {
data = new byte[dg.getLength()];
int dataPos = 0;
Datagram f = dg;
while (f != null) {
for (int i = dataPos, j =
Datagram.HEADER_SIZE; j < f.data.length; i++, j++)
data[i] = f.data[j];
dataPos += f.data.length -
Datagram.HEADER_SIZE;
f = f.nextFragment;
}
}
transportListener.messageArrivedEvent(dg.getTopicId(),
dg.getSenderId(), data);
}
}
// Resends the message if it was found in the cache
void resendDatagram(UDPNodeId requestor, int messageId, boolean broadcast)
throws InterruptedException, java.io.IOException {
slowDownSendRate();
if (broadcast) {
Datagram dg = broadcastStream.getFromSentCache(messageId);
if (dg != null) {
Log.log("[" + this +"] BROADCAST RESEND REQUEST
SERVICED: " + messageId);
send(dg, false);
return;
} else {
Log.log("[" + this +"] BROADCAST RESEND REQUEST NOT
SERVICED (not in cache): " + messageId);
}
} else {
Datagram dg = pointStream.getFromSentCache(messageId);
if (dg != null) {
Log.log("[" + this +"] POINT RESEND REQUEST SERVICED:
" + messageId);
send(requestor, dg, false);
return;
} else {
Log.log("[" + this +"] POINT RESEND REQUEST NOT
SERVICED (not in cache): " + messageId);
}
}
ResendErrorAdminDatagram error = new
ResendErrorAdminDatagram(messageId, broadcast);
adminSend(requestor, error);
}
// This is set when the transport is set in the cluster.
public void slowDownSendRate() {
stopSendsTill = System.currentTimeMillis() + 1000;
}
// Sends a datagram fragment chain over the network
synchronized void send(Datagram dg) throws InterruptedException,
java.io.IOException {
send(dg, true);
}
// Sends a datagram fragment chain over the network
synchronized void send(UDPNodeId dest, Datagram dg) throws
InterruptedException, java.io.IOException {
send(dest, dg, true);
}
// Sends a datagram fragment chain over the network
synchronized void send(Datagram dg, boolean setDatagramId) throws
InterruptedException, java.io.IOException {
int id = 0;
if (setDatagramId) {
id = broadcastStream.getNextDatagramId();
dg.setId(id);
}
broadcastStream.addToSentCache(dg);
// Send the fragment chain.
Log.log("[" + this +"] Sending Broadcast datagram: " + dg.getId());
while (dg != null) {
if (setDatagramId)
dg.setId(id);
DatagramPacket packet = new java.net.DatagramPacket(dg.data,
dg.data.length, clusterId.address, clusterId.port);
broadcastStream.socket.send(packet);
dg = dg.nextFragment;
}
}
// Sends a datagram fragment chain over the network
synchronized void send(UDPNodeId dest, Datagram dg, boolean setDatagramId)
throws InterruptedException, java.io.IOException {
int id = 0;
if (setDatagramId) {
id = broadcastStream.getNextDatagramId();
dg.setId(id);
}
pointStream.addToSentCache(dg);
Log.log("[" + this +"] Sending Point datagram: " + dg.getId());
// Send the fragment chain.
while (dg != null) {
if (setDatagramId)
dg.setId(id);
DatagramPacket packet = new java.net.DatagramPacket(dg.data,
dg.data.length, dest.address, dest.port);
pointStream.socket.send(packet);
dg = dg.nextFragment;
}
}
// Sends a resend request
void requestResend(UDPNodeId nodeId, Vector messageIds, boolean broadcast) {
int MAX_RESEND_ARRAY_SIZE = 100;
int msgs[] = new int[messageIds.size() < MAX_RESEND_ARRAY_SIZE ?
messageIds.size() : MAX_RESEND_ARRAY_SIZE];
for (int i = 0; i < msgs.length; i++) {
msgs[i] = ((Integer) messageIds.elementAt(i)).intValue();
}
Log.log("[" + this +"] SENDING RESEND REQUEST: " + messageIds);
ResendAdminDatagram m = new ResendAdminDatagram(msgs, broadcast);
try {
adminSend(nodeId, m);
} catch (Exception e) {
e.printStackTrace();
}
}
// Stops the transport
synchronized public void close() throws InterruptedException {
}
public NodeId getLocalNodeId() {
return nodeId;
}
// Builds a datagram fragment chain with the given data
// and then sends it over the network
public void send(NodeId dest, short channelId, byte data[], boolean droppable,
boolean keepOrder)
throws IOException, InterruptedException {
int length = data.length;
short fragmentId = 0;
byte msgFlags = 0;
if (droppable)
msgFlags = (byte) (msgFlags | Datagram.DROPPABLE_FLAG);
if (keepOrder)
msgFlags = (byte) (msgFlags | Datagram.KEEP_ORDER_FLAG);
Datagram firstFragment = null;
Datagram lastFragment = null;
// Check to see if the transmision rate has been slowed down.
long waitLeft = stopSendsTill - System.currentTimeMillis();
while (waitLeft > 0) {
Thread.sleep(waitLeft);
waitLeft = stopSendsTill - System.currentTimeMillis();
}
// while the whole message has not been proccesed
while (fragmentId * maxFragmentSize < length) {
Datagram dg;
// is what is left bigger than a fragment?
if ((length - (fragmentId * maxFragmentSize)) >
maxFragmentSize) {
// This a MAX size fragment
dg = new Datagram(length, fragmentId, nodeId,
channelId, msgFlags, data, fragmentId * maxFragmentSize, maxFragmentSize);
} else {
// This a not a MAX size fragment
dg =
new Datagram(
length,
fragmentId,
nodeId,
channelId,
msgFlags,
data,
fragmentId * maxFragmentSize,
length - (fragmentId *
maxFragmentSize));
}
fragmentId++;
// Chain the fragments together.
if (firstFragment == null) {
firstFragment = lastFragment = dg;
} else {
lastFragment.nextFragment = dg;
lastFragment = dg;
}
}
// Send the fragment chain.
if (dest == null)
send(firstFragment);
else
send((UDPNodeId) dest, firstFragment);
}
// This method is used to initialize is
// transport layer.
synchronized public void setProperties(Properties t) throws
InvalidConfigurationException, InvalidStateException {
if (started)
throw new InvalidStateException("Transport is allready
started");
String transportMode = t.getProperty("TransportMode");
String groupString = t.getProperty("Group");
String portString = t.getProperty("Port");
String nodeInterfaceString = t.getProperty("NodeInterface");
String nodePortString = t.getProperty("NodePort");
if (transportMode == null) {
throw new InvalidConfigurationException("TransportMode
property not set");
} else if (transportMode.equals("Multicast")) {
if (portString == null) {
throw new InvalidConfigurationException("Port property
not set");
}
if (groupString == null) {
throw new InvalidConfigurationException("Group
property not set");
}
try {
InetAddress group = InetAddress.getByName(groupString);
int port = Integer.parseInt(portString);
DatagramSocket pointSocket;
DatagramSocket adminSocket;
MulticastSocket broadcastSocket = new
MulticastSocket(port);
broadcastSocket.joinGroup(group);
if (nodeInterfaceString != null) {
nodeId.address =
InetAddress.getByName(nodeInterfaceString);
} else {
nodeId.address = InetAddress.getLocalHost();
}
broadcastSocket.setInterface(nodeId.address);
if (nodePortString != null) {
nodeId.port = Integer.parseInt(nodePortString);
pointSocket = new DatagramSocket(nodeId.port);
} else {
pointSocket = new DatagramSocket();
nodeId.port = pointSocket.getLocalPort();
}
adminSocket = new DatagramSocket();
nodeId.adminPort = adminSocket.getLocalPort();
Log.log("[" + this +"] Local NodeId: " + nodeId);
clusterId.address = group;
clusterId.port = port;
broadcastStream.socket = broadcastSocket;
pointStream.socket = pointSocket;
adminStream.socket = adminSocket;
} catch (IOException e) {
throw new InvalidConfigurationException("Ip group of
the Group property was invalid: " + e.getMessage());
}
} else {
throw new InvalidConfigurationException("TransportMode
property invalid");
}
}
// This is set when the transport is set in the cluster.
public void setTransportListener(TransportListener c) {
transportListener = c;
}
}
