ziver/hal
1
1
Fork 0
Code Pull requests Releases Activity

changed folder name

Browse source
This commit is contained in:
Ziver Koc 2011-02-15 19:40:02 +00:00
parent 4503531ec9
commit 80c6a52c69
73 changed files with 0 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

57
src/zutil/net/nio/NioClient.java Normal file
View file

@ -0,0 +1,57 @@
package zutil.net.nio;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import zutil.net.nio.message.Message;
import zutil.net.nio.message.type.ResponseRequestMessage;
import zutil.net.nio.response.ResponseEvent;
public class NioClient extends NioNetwork{
private SocketChannel serverSocket;
/**
* Creates a NioClient that connects to a server
*
* @param hostAddress The server address
* @param port The port to listen on
*/
public NioClient(InetAddress serverAddress, int port) throws IOException {
super(InetAddress.getLocalHost(), port, NetworkType.CLIENT);
serverSocket = initiateConnection(new InetSocketAddress(serverAddress, port));
Thread thread = new Thread(this);
thread.setDaemon(false);
thread.start();
}
protected Selector initSelector() throws IOException {
// Create a new selector
return SelectorProvider.provider().openSelector();
}
/**
* Sends a Message to the default server
*
* @param data The data to be sent
* @throws IOException Something got wrong
*/
public void send(Message data) throws IOException {
send(serverSocket, data);
}
/**
* This method is for the Client to send a message to the server
*
* @param handler The response handler
* @param data The data to send
* @throws IOException
*/
public void send(ResponseEvent handler, ResponseRequestMessage data) throws IOException {
send(serverSocket, handler, data);
}
}

452
src/zutil/net/nio/NioNetwork.java Normal file
View file

@ -0,0 +1,452 @@
package zutil.net.nio;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.logging.Logger;
import zutil.Encrypter;
import zutil.converters.Converter;
import zutil.io.DynamicByteArrayStream;
import zutil.io.MultiPrintStream;
import zutil.log.LogUtil;
import zutil.net.nio.message.type.ResponseRequestMessage;
import zutil.net.nio.message.type.SystemMessage;
import zutil.net.nio.response.ResponseEvent;
import zutil.net.nio.server.ChangeRequest;
import zutil.net.nio.server.ClientData;
import zutil.net.nio.worker.SystemWorker;
import zutil.net.nio.worker.Worker;
public abstract class NioNetwork implements Runnable {
private static Logger logger = LogUtil.getLogger();
public static enum NetworkType {SERVER, CLIENT};
private NetworkType type;
// The host:port combination to listen on
protected InetAddress address;
protected int port;
// The channel on which we'll accept connections
protected ServerSocketChannel serverChannel;
// The selector we'll be monitoring
private Selector selector;
// The buffer into which we'll read data when it's available
private ByteBuffer readBuffer = ByteBuffer.allocate(8192);
protected Worker worker;
protected SystemWorker systemWorker;
// This map contains all the clients that are conncted
protected Map<InetSocketAddress, ClientData> clients = new HashMap<InetSocketAddress, ClientData>();
// A list of PendingChange instances
private List<ChangeRequest> pendingChanges = new LinkedList<ChangeRequest>();
// Maps a SocketChannel to a list of ByteBuffer instances
private Map<SocketChannel, List<ByteBuffer>> pendingWriteData = new HashMap<SocketChannel, List<ByteBuffer>>();
private Map<SocketChannel, DynamicByteArrayStream> pendingReadData = new HashMap<SocketChannel, DynamicByteArrayStream>();
// The encrypter
private Encrypter encrypter;
/**
* Create a nio network class
*
* @param hostAddress The host address
* @param port The port
* @param type The type of network host
* @throws IOException
*/
public NioNetwork(InetAddress address, int port, NetworkType type) throws IOException {
this.port = port;
this.address = address;
this.type = type;
this.selector = initSelector();
this.systemWorker = new SystemWorker(this);
}
protected abstract Selector initSelector() throws IOException;
/**
* Sets the Worker for the network messages
*
* @param worker The worker that handles the incoming messages
*/
public void setDefaultWorker(Worker worker){
this.worker = worker;
}
/**
* Sets the encrypter to use in the network
*
* @param enc The encrypter to use or null fo no encryption
*/
public void setEncrypter(Encrypter enc){
encrypter = enc;
MultiPrintStream.out.println("Network Encryption "+
(encrypter != null ? "Enabled("+encrypter.getAlgorithm()+")" : "Disabled")+"!!");
}
public void send(SocketChannel socket, Object data) throws IOException{
send(socket, Converter.toBytes(data));
}
public void send(InetSocketAddress address, Object data) throws IOException{
send(address, Converter.toBytes(data));
}
public void send(InetSocketAddress address, byte[] data){
send(getSocketChannel(address), data);
}
public void send(SocketChannel socket, ResponseEvent handler, ResponseRequestMessage data) throws IOException {
// Register the response handler
systemWorker.addResponseHandler(handler, data);
queueSend(socket,Converter.toBytes(data));
}
/**
* This method sends data true the given socket
*
* @param socket The socket
* @param data The data to send
*/
public void send(SocketChannel socket, byte[] data) {
queueSend(socket,data);
}
/**
* Queues the message to be sent and wakeups the selector
*
* @param socket The socet to send the message thrue
* @param data The data to send
*/
protected void queueSend(SocketChannel socket, byte[] data){
logger.finest("Sending Queue...");
// And queue the data we want written
synchronized (pendingWriteData) {
List<ByteBuffer> queue = pendingWriteData.get(socket);
if (queue == null) {
queue = new ArrayList<ByteBuffer>();
pendingWriteData.put(socket, queue);
}
//encrypts
if(encrypter != null)
queue.add(ByteBuffer.wrap(encrypter.encrypt(data)));
else queue.add(ByteBuffer.wrap(data));
}
// Changing the key state to write
synchronized (pendingChanges) {
// Indicate we want the interest ops set changed
pendingChanges.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
}
logger.finest("selector.wakeup();");
// Finally, wake up our selecting thread so it can make the required changes
selector.wakeup();
}
public void run() {
logger.fine("NioNetwork Started!!!");
while (true) {
try {
// Process any pending changes
synchronized (pendingChanges) {
Iterator<ChangeRequest> changes = pendingChanges.iterator();
while (changes.hasNext()) {
ChangeRequest change = changes.next();
switch (change.type) {
case ChangeRequest.CHANGEOPS:
SelectionKey key = change.socket.keyFor(selector);
key.interestOps(change.ops);
logger.finest("change.ops "+change.ops);
break;
case ChangeRequest.REGISTER:
change.socket.register(selector, change.ops);
logger.finest("register socket ");
break;
}
}
pendingChanges.clear();
}
// Wait for an event one of the registered channels
selector.select();
logger.finest("selector is awake");
// Iterate over the set of keys for which events are available
Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
while (selectedKeys.hasNext()) {
SelectionKey key = (SelectionKey) selectedKeys.next();
selectedKeys.remove();
logger.finest("KeyOP: "+key.interestOps()+" isAcceptable: "+SelectionKey.OP_ACCEPT+" isConnectable: "+SelectionKey.OP_CONNECT+" isWritable: "+SelectionKey.OP_WRITE+" isReadable: "+SelectionKey.OP_READ);
if (key.isValid()) {
// Check what event is available and deal with it
if (key.isAcceptable()) {
logger.finest("Accepting Connection!!");
accept(key);
}
else if (key.isConnectable()) {
logger.finest("Finnishing Connection!!");
finishConnection(key);
}
else if (key.isWritable()) {
logger.finest("Writing");
write(key);
}
else if (key.isReadable()) {
logger.finest("Reading");
read(key);
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* Server
*/
private void accept(SelectionKey key) throws IOException {
// For an accept to be pending the channel must be a server socket channel.
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
// Accept the connection and make it non-blocking
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.socket().setReuseAddress(true);
socketChannel.configureBlocking(false);
// Register the new SocketChannel with our Selector, indicating
// we'd like to be notified when there's data waiting to be read
socketChannel.register(selector, SelectionKey.OP_READ);
// adds the client to the clients list
InetSocketAddress remoteAdr = (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
if(!clients.containsValue(remoteAdr)){
clients.put(remoteAdr, new ClientData(socketChannel));
logger.fine("New Connection("+remoteAdr+")!!! Count: "+clients.size());
}
}
/**
* Client and Server
*/
private void read(SelectionKey key) throws IOException {
SocketChannel socketChannel = (SocketChannel) key.channel();
InetSocketAddress remoteAdr = (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
// Clear out our read buffer so it's ready for new data
readBuffer.clear();
// Attempt to read off the channel
int numRead;
try {
numRead = socketChannel.read(readBuffer);
} catch (IOException e) {
// The remote forcibly closed the connection, cancel
// the selection key and close the channel.
key.cancel();
socketChannel.close();
clients.remove(remoteAdr);
pendingReadData.remove(socketChannel);
pendingWriteData.remove(socketChannel);
logger.fine("Connection Forced Close("+remoteAdr+")!!! Connection Count: "+clients.size());
if(type == NetworkType.CLIENT)
throw new IOException("Server Closed The Connection!!!");
return;
}
if (numRead == -1) {
// Remote entity shut the socket down cleanly. Do the
// same from our end and cancel the channel.
key.channel().close();
key.cancel();
clients.remove(remoteAdr);
pendingReadData.remove(socketChannel);
pendingWriteData.remove(socketChannel);
logger.fine("Connection Close("+remoteAdr+")!!! Connection Count: "+clients.size());
if(type == NetworkType.CLIENT)
throw new IOException("Server Closed The Connection!!!");
return;
}
// Make a correctly sized copy of the data before handing it
// to the client
byte[] rspByteData = new byte[numRead];
System.arraycopy(readBuffer.array(), 0, rspByteData, 0, numRead);
if(encrypter != null)// Encryption
rspByteData = encrypter.decrypt(rspByteData);
// Message Count 1m: 36750
// Message Count 1s: 612
if(!pendingReadData.containsKey(socketChannel)){
pendingReadData.put(socketChannel, new DynamicByteArrayStream());
}
DynamicByteArrayStream dynBuf = pendingReadData.get(socketChannel);
dynBuf.add(rspByteData);
Object rspData = null;
try{
//rspData = Converter.toObject(rspByteData);
rspData = Converter.toObject(dynBuf);
handleRecivedMessage(socketChannel, rspData);
dynBuf.clear();
}catch(Exception e){
e.printStackTrace();
dynBuf.reset();
}
}
/**
* Client and Server
*/
private void write(SelectionKey key) throws IOException {
SocketChannel socketChannel = (SocketChannel) key.channel();
synchronized (pendingWriteData) {
List<ByteBuffer> queue = pendingWriteData.get(socketChannel);
if(queue == null){
queue = new ArrayList<ByteBuffer>();
pendingWriteData.put(socketChannel, queue);
}
// Write until there's not more data ...
while (!queue.isEmpty()) {
ByteBuffer buf = queue.get(0);
socketChannel.write(buf);
if (buf.remaining() > 0) {
// ... or the socket's buffer fills up
logger.finest("Write Buffer Full!!");
break;
}
queue.remove(0);
}
if (queue.isEmpty()) {
// We wrote away all data, so we're no longer interested
// in writing on this socket. Switch back to waiting for
// data.
logger.finest("No more Data to write!!");
key.interestOps(SelectionKey.OP_READ);
}
}
}
private void handleRecivedMessage(SocketChannel socketChannel, Object rspData){
logger.finer("Handling incomming message...");
if(rspData instanceof SystemMessage){
if(systemWorker != null){
logger.finest("System Message!!!");
systemWorker.processData(this, socketChannel, rspData);
}
else{
logger.finer("Unhandled System Message!!!");
}
}
else{
// Hand the data off to our worker thread
if(worker != null){
logger.finest("Worker Message!!!");
worker.processData(this, socketChannel, rspData);
}
else{
logger.fine("Unhandled Worker Message!!!");
}
}
}
/**
* Initializes a socket to a server
*/
protected SocketChannel initiateConnection(InetSocketAddress address) throws IOException {
// Create a non-blocking socket channel
SocketChannel socketChannel = SocketChannel.open();
socketChannel.socket().setReuseAddress(true);
socketChannel.configureBlocking(false);
logger.fine("Connecting to: "+address);
// Kick off connection establishment
socketChannel.connect(address);
// Queue a channel registration since the caller is not the
// selecting thread. As part of the registration we'll register
// an interest in connection events. These are raised when a channel
// is ready to complete connection establishment.
synchronized(this.pendingChanges) {
pendingChanges.add(new ChangeRequest(socketChannel, ChangeRequest.REGISTER, SelectionKey.OP_CONNECT));
}
return socketChannel;
}
protected SocketChannel getSocketChannel(InetSocketAddress address){
return clients.get(address).getSocketChannel();
}
/**
* Client
*/
private void finishConnection(SelectionKey key){
SocketChannel socketChannel = (SocketChannel) key.channel();
// Finish the connection. If the connection operation failed
// this will raise an IOException.
try {
socketChannel.finishConnect();
} catch (IOException e) {
// Cancel the channel's registration with our selector
e.printStackTrace();
key.cancel();
return;
}
// Register an interest in writing on this channel
key.interestOps(SelectionKey.OP_WRITE);
}
/**
* Client
* @throws IOException
*/
protected void closeConnection(SocketChannel socketChannel) throws IOException{
socketChannel.close();
socketChannel.keyFor(selector).cancel();
}
/**
* Client
* @throws IOException
*/
protected void closeConnection(InetSocketAddress address) throws IOException{
closeConnection(getSocketChannel(address));
}
/*
public void close() throws IOException{
if(serverChannel != null){
serverChannel.close();
serverChannel.keyFor(selector).cancel();
}
selector.close();
}*/
public NetworkType getType(){
return type;
}
}

68
src/zutil/net/nio/NioServer.java Normal file
View file

@ -0,0 +1,68 @@
package zutil.net.nio;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
public class NioServer extends NioNetwork{
/**
* Creates a NioServer object which listens on localhost
*
* @param port The port to listen to
*/
public NioServer(int port) throws IOException {
this(null, port);
}
/**
* Creates a NioServer object which listens to a specific address
*
* @param address The address to listen to
* @param port The port to listen to
*/
public NioServer(InetAddress address, int port) throws IOException {
super(address, port, NetworkType.SERVER);
new Thread(this).start();
}
protected Selector initSelector() throws IOException {
// Create a new selector
Selector socketSelector = SelectorProvider.provider().openSelector();
// Create a new non-blocking server socket channel
serverChannel = ServerSocketChannel.open();
serverChannel.socket().setReuseAddress(true);
serverChannel.configureBlocking(false);
// Bind the server socket to the specified address and port
InetSocketAddress isa = new InetSocketAddress(address, port);
serverChannel.socket().bind(isa);
// Register the server socket channel, indicating an interest in
// accepting new connections
serverChannel.register(socketSelector, SelectionKey.OP_ACCEPT);
return socketSelector;
}
/**
* Broadcasts the message to all the connected clients
*
* @param data The data to broadcast
*/
public void broadcast(byte[] data){
synchronized(clients){
Iterator<InetSocketAddress> it = clients.keySet().iterator();
while(it.hasNext()){
send(it.next(), data);
}
}
}
}

45
src/zutil/net/nio/message/ChatMessage.java Normal file
View file

@ -0,0 +1,45 @@
package zutil.net.nio.message;
public class ChatMessage extends Message{
private static final long serialVersionUID = 1L;
public static enum ChatMessageType {REGISTER, UNREGISTER, MESSAGE};
public ChatMessageType type;
public String msg;
public String room;
/**
* Registers the user to the main chat
*
* @param name Name of user
*/
public ChatMessage(){
this("", "", ChatMessageType.REGISTER);
}
/**
* Registers the user to the given room
*
* @param room The room to register to
*/
public ChatMessage(String room){
this("", room, ChatMessageType.REGISTER);
}
/**
* Sends a message to the given room
*
* @param msg The message
* @param room The room
*/
public ChatMessage(String msg, String room){
this(msg, room, ChatMessageType.MESSAGE);
}
public ChatMessage(String msg, String room, ChatMessageType type){
this.msg = msg;
this.room = room;
this.type = type;
}
}

31
src/zutil/net/nio/message/GraphicsSyncMessage.java Normal file
View file

@ -0,0 +1,31 @@
package zutil.net.nio.message;
public class GraphicsSyncMessage extends SyncMessage{
private static final long serialVersionUID = 1L;
public float locX;
public float locY;
public float locZ;
public float rotX;
public float rotY;
public float rotZ;
public float rotW;
public GraphicsSyncMessage(String id){
this.type = MessageType.SYNC;
this.id = id;
}
public boolean equals(Object obj){
if(obj instanceof GraphicsSyncMessage){
GraphicsSyncMessage tmp = (GraphicsSyncMessage)obj;
return (tmp.locX == locX && tmp.locY == locY &&
tmp.locZ == locZ && tmp.rotX == rotX &&
tmp.rotY == rotY && tmp.rotZ == rotZ &&
tmp.rotW == rotW);
}
return false;
}
}

84
src/zutil/net/nio/message/GridMessage.java Normal file
View file

@ -0,0 +1,84 @@
package zutil.net.nio.message;
public class GridMessage<T> extends Message{
private static final long serialVersionUID = 1L;
// Client type messages
/** Computation job return right answer **/
public static final int COMP_SUCCESSFUL = 1; //
/** Initial static data **/
public static final int COMP_INCORRECT = 2; //
/** Computation job return wrong answer **/
public static final int COMP_ERROR = 3; //
/** There was an error computing **/
public static final int REGISTER = 4; //
/** Register at the server **/
public static final int UNREGISTER = 5; //
/** Request new computation data **/
public static final int NEW_DATA = 6; //
// Server type messages
/** Sending initial static data **/
public static final int INIT_DATA = 100;
/** Sending new dynamic data **/
public static final int COMP_DATA = 101;
private int type;
private int jobID;
private T data;
/**
* Creates a new GridMessage
*
* @param type is the type of message
* @param jobID is the id of the job
*/
public GridMessage(int type){
this(type, 0, null);
}
/**
* Creates a new GridMessage
*
* @param type is the type of message
* @param jobID is the id of the job
*/
public GridMessage(int type, int jobID){
this(type, jobID, null);
}
/**
* Creates a new GridMessage
*
* @param type is the type of message
* @param jobID is the id of the job
* @param data is the data to send with this message
*/
public GridMessage(int type, int jobID, T data){
this.type = type;
this.jobID = jobID;
this.data = data;
}
/**
* @return the type of message
*/
public int messageType(){
return type;
}
/**
* @return the job id for this message
*/
public int getJobQueueID(){
return jobID;
}
/**
* @return the data in this message, may not always carry any data.
*/
public T getData(){
return data;
}
}

16
src/zutil/net/nio/message/KeepAliveMessage.java Normal file
View file

@ -0,0 +1,16 @@
package zutil.net.nio.message;
import zutil.net.nio.message.type.SystemMessage;
/**
* Tells the destination that the
* source is still online
*
* @author Ziver
*
*/
public class KeepAliveMessage extends Message implements SystemMessage{
private static final long serialVersionUID = 1L;
}

9
src/zutil/net/nio/message/Message.java Normal file
View file

@ -0,0 +1,9 @@
package zutil.net.nio.message;
import java.io.Serializable;
public class Message implements Serializable{
private static final long serialVersionUID = 1L;
}

34
src/zutil/net/nio/message/StringMessage.java Normal file
View file

@ -0,0 +1,34 @@
package zutil.net.nio.message;
import zutil.net.nio.message.type.EchoMessage;
import zutil.net.nio.message.type.ResponseRequestMessage;
public class StringMessage extends EchoMessage implements ResponseRequestMessage{
private static final long serialVersionUID = 1L;
private double responseId;
private String msg;
public StringMessage(String msg){
this.msg = msg;
responseId = Math.random();
}
public String getString(){
return msg;
}
public void setString(String msg){
this.msg = msg;
}
public String toString(){
return getString();
}
public double getResponseId() {
return responseId;
}
}

13
src/zutil/net/nio/message/SyncMessage.java Normal file
View file

@ -0,0 +1,13 @@
package zutil.net.nio.message;
import zutil.net.nio.message.type.SystemMessage;
public class SyncMessage extends Message implements SystemMessage{
private static final long serialVersionUID = 1L;
public static enum MessageType { REQUEST_ID, NEW, REMOVE, SYNC };
// type of message
public MessageType type;
// id of the Object
public String id;
}

34
src/zutil/net/nio/message/type/EchoMessage.java Normal file
View file

@ -0,0 +1,34 @@
package zutil.net.nio.message.type;
import zutil.net.nio.message.Message;
/**
* The reciver will echo out this message to the sender
*
* @author Ziver
*/
public abstract class EchoMessage extends Message implements SystemMessage{
private static final long serialVersionUID = 1L;
private boolean echo;
public EchoMessage(){
echo = true;
}
/**
* This method returns if the message should be echoed
* @return If the message should be echoed
*/
public boolean echo() {
return echo;
}
/**
* Called by the reciver to disable looping of the message
*
*/
public void recived() {
echo = false;
}
}

18
src/zutil/net/nio/message/type/ResponseRequestMessage.java Normal file
View file

@ -0,0 +1,18 @@
package zutil.net.nio.message.type;
/**
* This interface means that the sender
* wants a reply from the destination
*
* @author Ziver
*
*/
public interface ResponseRequestMessage {
/**
* The id of the response to identify the response event
* @return Response id
*/
public double getResponseId();
}

12
src/zutil/net/nio/message/type/SystemMessage.java Normal file
View file

@ -0,0 +1,12 @@
package zutil.net.nio.message.type;
/**
* A message that implements this will be
* handeld internaly by the network engine
*
* @author Ziver
*
*/
public interface SystemMessage {
}

12
src/zutil/net/nio/response/PrintRsp.java Normal file
View file

@ -0,0 +1,12 @@
package zutil.net.nio.response;
import zutil.io.MultiPrintStream;
public class PrintRsp extends ResponseEvent{
@Override
protected void responseEvent(Object rsp) {
MultiPrintStream.out.println(rsp);
}
}

43
src/zutil/net/nio/response/ResponseEvent.java Normal file
View file

@ -0,0 +1,43 @@
package zutil.net.nio.response;
public abstract class ResponseEvent {
private Object rsp = null;
public synchronized boolean handleResponse(Object rsp) {
this.rsp = rsp;
notify();
return true;
}
/**
* Blocks the Thread until there is a response
*/
public synchronized void waitForResponse() {
while(!gotResponse()) {
try {
this.wait();
} catch (InterruptedException e) {}
}
responseEvent(rsp);
}
/**
* Handles the response
*/
public void handleResponse(){
if(gotResponse()){
responseEvent(rsp);
}
}
/**
* @return If there is an response
*/
public boolean gotResponse(){
return (rsp != null);
}
protected abstract void responseEvent(Object rsp);
}

41
src/zutil/net/nio/response/ResponseHandler.java Normal file
View file

@ -0,0 +1,41 @@
package zutil.net.nio.response;
import java.util.LinkedList;
import java.util.List;
public abstract class ResponseHandler implements Runnable{
private List<ResponseEvent> queue = new LinkedList<ResponseEvent>();
public ResponseHandler(){
}
public synchronized void addResponseEvent(ResponseEvent re){
queue.add(re);
notify();
}
public synchronized void removeResponseEvent(ResponseEvent re){
queue.remove(re);
}
public void run() {
while(true) {
try {
this.wait();
} catch (InterruptedException e) {}
update();
}
}
public synchronized void update(){
while(!queue.isEmpty()){
queue.get(0).handleResponse();
if(queue.get(0).gotResponse()){
queue.remove(0);
}
}
}
}

18
src/zutil/net/nio/server/ChangeRequest.java Normal file
View file

@ -0,0 +1,18 @@
package zutil.net.nio.server;
import java.nio.channels.SocketChannel;
public class ChangeRequest {
public static final int REGISTER = 1;
public static final int CHANGEOPS = 2;
public SocketChannel socket;
public int type;
public int ops;
public ChangeRequest(SocketChannel socket, int type, int ops) {
this.socket = socket;
this.type = type;
this.ops = ops;
}
}

29
src/zutil/net/nio/server/ClientData.java Normal file
View file

@ -0,0 +1,29 @@
package zutil.net.nio.server;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
public class ClientData {
private SocketChannel socketChannel;
private long lastMessageReceived;
public ClientData(SocketChannel socketChannel){
this.socketChannel = socketChannel;
}
public SocketChannel getSocketChannel(){
return socketChannel;
}
public InetSocketAddress getAddress(){
return (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
}
public void setLastMessageReceived(long time){
lastMessageReceived = time;
}
public long getLastMessageReceived(){
return lastMessageReceived;
}
}

25
src/zutil/net/nio/service/NetworkService.java Normal file
View file

@ -0,0 +1,25 @@
package zutil.net.nio.service;
import java.nio.channels.SocketChannel;
import zutil.net.nio.NioNetwork;
import zutil.net.nio.message.Message;
public abstract class NetworkService {
protected static NetworkService instance;
protected NioNetwork nio;
public NetworkService(NioNetwork nio){
instance = this;
this.nio = nio;
}
public abstract void handleMessage(Message message, SocketChannel socket);
/**
* @return A instance of this class
*/
public static NetworkService getInstance(){
return instance;
}
}

10
src/zutil/net/nio/service/chat/ChatListener.java Normal file
View file

@ -0,0 +1,10 @@
package zutil.net.nio.service.chat;
/**
* Tis is a listener class for new chat messages
* @author Ziver
*
*/
public interface ChatListener {
public void messageAction(String msg, String room);
}

133
src/zutil/net/nio/service/chat/ChatService.java Normal file
View file

@ -0,0 +1,133 @@
package zutil.net.nio.service.chat;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.logging.Logger;
import zutil.log.LogUtil;
import zutil.net.nio.NioNetwork;
import zutil.net.nio.message.ChatMessage;
import zutil.net.nio.message.Message;
import zutil.net.nio.service.NetworkService;
/**
* A simple chat service with users and rooms
*
* @author Ziver
*/
public class ChatService extends NetworkService{
private static Logger logger = LogUtil.getLogger();
private HashMap<String,LinkedList<SocketChannel>> rooms;
private ChatListener listener;
public ChatService(NioNetwork nio){
super(nio);
rooms = new HashMap<String,LinkedList<SocketChannel>>();
}
@Override
public void handleMessage(Message message, SocketChannel socket) {
try {
// New message
if(message instanceof ChatMessage){
ChatMessage chatmessage = (ChatMessage)message;
//is this a new message
if(chatmessage.type == ChatMessage.ChatMessageType.MESSAGE){
// Is this the server
if(nio.getType() == NioNetwork.NetworkType.SERVER){
if(rooms.containsKey(chatmessage.room)){
LinkedList<SocketChannel> tmpList = rooms.get(chatmessage.room);
// Broadcast the message
for(SocketChannel s : tmpList){
if(s.isConnected()){
nio.send(s, chatmessage);
}
else{
unRegisterUser(chatmessage.room, s);
}
}
}
}
logger.finer("New Chat Message: "+chatmessage.msg);
listener.messageAction(chatmessage.msg, chatmessage.room);
}
// register to a room
else if(chatmessage.type == ChatMessage.ChatMessageType.REGISTER){
registerUser(chatmessage.room, socket);
}
// unregister to a room
else if(chatmessage.type == ChatMessage.ChatMessageType.UNREGISTER){
unRegisterUser(chatmessage.room, socket);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* Registers a user to the main room
*
* @param socket The socket to the user
*/
public void registerUser(SocketChannel socket){
registerUser("", socket);
}
/**
* Registers the given user to a specific room
*
* @param room The room
* @param socket The socket to the user
*/
public void registerUser(String room, SocketChannel socket){
addRoom(room);
logger.fine("New Chat User: "+socket);
rooms.get(room).add(socket);
}
/**
* Unregisters a user from a room and removes the room if its empty
*
* @param room The room
* @param socket The socket to the user
*/
public void unRegisterUser(String room, SocketChannel socket){
if(rooms.containsKey(room)){
logger.fine("Remove Chat User: "+socket);
rooms.get(room).remove(socket);
removeRoom(room);
}
}
/**
* Adds a room into the list
*
* @param room The name of the room
*/
public void addRoom(String room){
if(!rooms.containsKey(room)){
logger.fine("New Chat Room: "+room);
rooms.put(room, new LinkedList<SocketChannel>());
}
}
/**
* Removes the given room if its empty
*
* @param room The room
*/
public void removeRoom(String room){
if(rooms.get(room).isEmpty()){
logger.fine("Remove Chat Room: "+room);
rooms.remove(room);
}
}
public static ChatService getInstance(){
return (ChatService)instance;
}
}

28
src/zutil/net/nio/service/sync/ObjectSync.java Normal file
View file

@ -0,0 +1,28 @@
package zutil.net.nio.service.sync;
import zutil.net.nio.message.SyncMessage;
public abstract class ObjectSync {
public String id;
public ObjectSync(String id){
this.id = id;
}
/**
* Sends sync message if the object has bean changed
*/
public abstract void sendSync();
/**
* Applies the SyncMessage to the object
* @param message
* @param object
*/
public abstract void syncObject(SyncMessage message);
/**
* Called when the object is removed from the sync list
*/
public abstract void remove();
}

60
src/zutil/net/nio/service/sync/SyncService.java Normal file
View file

@ -0,0 +1,60 @@
package zutil.net.nio.service.sync;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.logging.Logger;
import zutil.log.LogUtil;
import zutil.net.nio.NioNetwork;
import zutil.net.nio.message.Message;
import zutil.net.nio.message.SyncMessage;
import zutil.net.nio.service.NetworkService;
public class SyncService extends NetworkService{
private static Logger logger = LogUtil.getLogger();
// list of objects to sync
private HashMap<String, ObjectSync> sync;
public SyncService(NioNetwork nio){
super(nio);
sync = new HashMap<String, ObjectSync>();
}
/**
* Adds a SyncObject to the sync list
* @param os The object to sync
*/
public void addSyncObject(ObjectSync os){
sync.put(os.id, os);
logger.fine("New Sync object: "+os);
}
public void handleMessage(Message message, SocketChannel socket){
if(message instanceof SyncMessage){
SyncMessage syncMessage = (SyncMessage)message;
if(syncMessage.type == SyncMessage.MessageType.SYNC){
ObjectSync obj = sync.get(syncMessage.id);
if(obj != null){
logger.finer("Syncing Message...");
obj.syncObject(syncMessage);
}
}
else if(syncMessage.type == SyncMessage.MessageType.REMOVE){
sync.remove(syncMessage.id).remove();
}
}
}
/**
* Syncs all the objects whit the server
*/
public void sync(){
for(String id : sync.keySet()){
sync.get(id).sendSync();
}
}
public static SyncService getInstance(){
return (SyncService)instance;
}
}

22
src/zutil/net/nio/worker/EchoWorker.java Normal file
View file

@ -0,0 +1,22 @@
package zutil.net.nio.worker;
import java.io.IOException;
import zutil.io.MultiPrintStream;
public class EchoWorker extends ThreadedEventWorker {
@Override
public void messageEvent(WorkerDataEvent dataEvent) {
try {
// Return to sender
MultiPrintStream.out.println("Recived Msg: "+dataEvent.data);
dataEvent.network.send(dataEvent.socket, dataEvent.data);
} catch (IOException e) {
e.printStackTrace();
}
}
}

129
src/zutil/net/nio/worker/SystemWorker.java Normal file
View file

@ -0,0 +1,129 @@
package zutil.net.nio.worker;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;
import zutil.log.LogUtil;
import zutil.net.nio.NioNetwork;
import zutil.net.nio.message.ChatMessage;
import zutil.net.nio.message.Message;
import zutil.net.nio.message.SyncMessage;
import zutil.net.nio.message.type.EchoMessage;
import zutil.net.nio.message.type.ResponseRequestMessage;
import zutil.net.nio.response.ResponseEvent;
import zutil.net.nio.service.NetworkService;
import zutil.net.nio.service.chat.ChatService;
import zutil.net.nio.service.sync.SyncService;
public class SystemWorker extends ThreadedEventWorker {
private static Logger logger = LogUtil.getLogger();
private NioNetwork nio;
// Maps a SocketChannel to a RspHandler
private Map<Double, ResponseEvent> rspEvents = new HashMap<Double, ResponseEvent>();
// Difren services listening on specific messages
private Map<Class<?>, NetworkService> services = new HashMap<Class<?>, NetworkService>();
/**
* Creates a new SystemWorker
* @param nio The Network
*/
public SystemWorker(NioNetwork nio){
this.nio = nio;
}
@Override
public void messageEvent(WorkerDataEvent event) {
try {
logger.finer("System Message: "+event.data.getClass().getName());
if(event.data instanceof Message){
if(event.data instanceof EchoMessage && ((EchoMessage)event.data).echo()){
// Echos back the recived message
((EchoMessage)event.data).recived();
logger.finer("Echoing Message: "+event.data);
nio.send(event.socket, event.data);
}
else if(event.data instanceof ResponseRequestMessage &&
rspEvents.get(((ResponseRequestMessage)event.data).getResponseId()) != null){
// Handle the response
handleResponse(((ResponseRequestMessage)event.data).getResponseId(), event.data);
logger.finer("Response Request Message: "+event.data);
}
else{
//Services
if(services.containsKey(event.data.getClass()) ||
!services.containsKey(event.data.getClass()) && defaultServices(event.data)){
services.get(event.data.getClass()).handleMessage((Message)event.data, event.socket);
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* Registers a Service to a specific message
*
* @param c The Message class
* @param ns The service
*/
public void registerService(Class<?> c, NetworkService ns){
services.put(c, ns);
}
/**
* Unregisters a service
*
* @param c The class
*/
public void unregisterService(Class<?> c){
services.remove(c);
}
/**
* Connects a ResponseHandler to a specific message
* @param handler The Handler
* @param data The Message
*/
public void addResponseHandler(ResponseEvent handler, ResponseRequestMessage data){
rspEvents.put(data.getResponseId(), handler);
}
/**
* Client And Server ResponseEvent
*/
private void handleResponse(double responseId, Object rspData){
// Look up the handler for this channel
ResponseEvent handler = rspEvents.get(responseId);
// And pass the response to it
handler.handleResponse(rspData);
rspEvents.remove(responseId);
}
/**
* Registers the default services in the engin e
* if the message needs one of them
*
* @param o The message
*/
private boolean defaultServices(Object o){
if(o instanceof SyncMessage){
if(SyncService.getInstance() == null)
registerService(o.getClass(), new SyncService(nio));
else
registerService(o.getClass(), SyncService.getInstance());
return true;
}
else if(o instanceof ChatMessage){
if(ChatService.getInstance() == null)
registerService(o.getClass(), new ChatService(nio));
else
registerService(o.getClass(), ChatService.getInstance());
return true;
}
return false;
}
}

34
src/zutil/net/nio/worker/ThreadedEventWorker.java Normal file
View file

@ -0,0 +1,34 @@
package zutil.net.nio.worker;
public abstract class ThreadedEventWorker extends Worker{
private Thread thread;
public ThreadedEventWorker(){
thread = new Thread(this);
thread.start();
}
public void update() {
WorkerDataEvent dataEvent;
while(true) {
try{
// Wait for data to become available
synchronized(getEventQueue()) {
while(getEventQueue().isEmpty()) {
try {
getEventQueue().wait();
} catch (InterruptedException e) {}
}
dataEvent = (WorkerDataEvent) getEventQueue().remove(0);
}
messageEvent(dataEvent);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public abstract void messageEvent(WorkerDataEvent e);
}

52
src/zutil/net/nio/worker/Worker.java Normal file
View file

@ -0,0 +1,52 @@
package zutil.net.nio.worker;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;
import zutil.net.nio.NioNetwork;
public abstract class Worker implements Runnable {
private LinkedList<WorkerDataEvent> queue = new LinkedList<WorkerDataEvent>();
public void processData(NioNetwork server, SocketChannel socket, Object data) {
synchronized(queue) {
queue.add(new WorkerDataEvent(server, socket, data));
queue.notify();
}
}
/**
* @return The event queue
*/
protected List<WorkerDataEvent> getEventQueue(){
return queue;
}
/**
* @return If there is a event in the queue
*/
protected boolean hasEvent(){
return !queue.isEmpty();
}
/**
* Polls a event from the list or waits until there is a event
* @return The next event
*/
protected WorkerDataEvent pollEvent(){
while(queue.isEmpty()) {
try {
this.wait();
} catch (InterruptedException e) {}
}
return queue.poll();
}
public void run(){
update();
}
public abstract void update();
}

18
src/zutil/net/nio/worker/WorkerDataEvent.java Normal file
View file

@ -0,0 +1,18 @@
package zutil.net.nio.worker;
import java.nio.channels.SocketChannel;
import zutil.net.nio.NioNetwork;
public class WorkerDataEvent {
public NioNetwork network;
public SocketChannel socket;
public Object data;
public WorkerDataEvent(NioNetwork server, SocketChannel socket, Object data) {
this.network = server;
this.socket = socket;
this.data = data;
}
}

114
src/zutil/net/nio/worker/grid/GridClient.java Normal file
View file

@ -0,0 +1,114 @@
package zutil.net.nio.worker.grid;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Queue;
import zutil.io.MultiPrintStream;
import zutil.net.nio.NioClient;
import zutil.net.nio.message.GridMessage;
import zutil.net.nio.worker.ThreadedEventWorker;
import zutil.net.nio.worker.WorkerDataEvent;
/**
* This class is the client part of the grid.
* It connects to a grid server and requests new job.
* And then sends back the result to the server.
*
* @author Ziver
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public class GridClient extends ThreadedEventWorker {
private static LinkedList<GridJob> jobQueue;
private static GridThread thread;
private static NioClient network;
/**
* Creates a new GridClient object and registers itself at the server
* and sets itself as a worker in NioClient
*
* @param thread the Thread interface to run for the jobs
* @param network the NioClient to use to communicate to the server
*/
public GridClient(GridThread thread, NioClient network){
jobQueue = new LinkedList<GridJob>();
GridClient.thread = thread;
GridClient.network = network;
}
/**
* Starts up the client and a couple of GridThreads.
* And registers itself as a worker in NioClient
* @throws IOException
*/
public void initiate() throws IOException{
network.setDefaultWorker(this);
network.send(new GridMessage(GridMessage.REGISTER));
for(int i=0; i<Runtime.getRuntime().availableProcessors() ;i++){
Thread t = new Thread(thread);
t.start();
}
}
@Override
public void messageEvent(WorkerDataEvent e) {
// ignores other messages than GridMessage
if(e.data instanceof GridMessage){
GridMessage msg = (GridMessage)e.data;
switch(msg.messageType()){
// Receive data from Server
case GridMessage.INIT_DATA:
thread.setInitData(msg.getData());
break;
case GridMessage.COMP_DATA:
jobQueue.add(new GridJob(msg.getJobQueueID(), (Queue)msg.getData()));
break;
}
}
}
/**
* Register whit the server that the job is done
*
* @param jobID is the job id
* @param correct if the answer was right
* @param result the result of the computation
* @throws IOException
*/
public static void jobDone(int jobID, boolean correct, Object result) throws IOException{
if(correct)
network.send(new GridMessage(GridMessage.COMP_SUCCESSFUL, jobID, result));
else
network.send(new GridMessage(GridMessage.COMP_INCORRECT, jobID, result));
}
/**
* Registers with the server that there was an
* error when computing this job
*
* @param jobID is the job id
*/
public static void jobError(int jobID){
try{
network.send(new GridMessage(GridMessage.COMP_SUCCESSFUL, jobID));
}catch(Exception e){e.printStackTrace();}
}
/**
* @return a new job to compute
* @throws IOException
*/
public static synchronized GridJob getNextJob() throws IOException{
if(jobQueue.isEmpty()){
network.send(new GridMessage(GridMessage.NEW_DATA));
while(jobQueue.isEmpty()){
try{Thread.sleep(100);}catch(Exception e){}
}
}
MultiPrintStream.out.println("Starting job");
return jobQueue.poll();
}
}

22
src/zutil/net/nio/worker/grid/GridJob.java Normal file
View file

@ -0,0 +1,22 @@
package zutil.net.nio.worker.grid;
/**
* A internal class for handling the jobs
*
* @author Ziver
*/
public class GridJob{
public int jobID;
public Object job;
public long timestamp;
public GridJob(int jobID, Object job){
this.jobID = jobID;
this.job = job;
renewTimeStamp();
}
public void renewTimeStamp(){
timestamp = System.currentTimeMillis();
}
}

18
src/zutil/net/nio/worker/grid/GridJobGenerator.java Normal file
View file

@ -0,0 +1,18 @@
package zutil.net.nio.worker.grid;
/**
* Generates new jobs for the grid to compute
*
* @author Ziver
*/
public interface GridJobGenerator<T> {
/**
* @return static and final values that do not change for every job
*/
public Object initValues();
/**
* @return a new generated job
*/
public T generateJob();
}

11
src/zutil/net/nio/worker/grid/GridResultHandler.java Normal file
View file

@ -0,0 +1,11 @@
package zutil.net.nio.worker.grid;
/**
* Handles the incoming results from the grid
*
* @author Ziver
*/
public interface GridResultHandler<T> {
public void resultEvent(int jobID, boolean correct, T result);
}

111
src/zutil/net/nio/worker/grid/GridServerWorker.java Normal file
View file

@ -0,0 +1,111 @@
package zutil.net.nio.worker.grid;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Queue;
import zutil.net.nio.message.GridMessage;
import zutil.net.nio.worker.ThreadedEventWorker;
import zutil.net.nio.worker.WorkerDataEvent;
/**
* Implements a simple network computing server
*
* @author Ziver
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public class GridServerWorker extends ThreadedEventWorker{
// Job timeout after 30 min
public static int JOB_TIMEOUT = 1000*60*30;
private HashMap<Integer, GridJob> jobs; // contains all the ongoing jobs
private Queue<GridJob> reSendjobQueue; // Contains all the jobs that will be recalculated
private GridJobGenerator jobGenerator; // The job generator
private GridResultHandler resHandler;
private int nextJobID;
public GridServerWorker(GridResultHandler resHandler, GridJobGenerator jobGenerator){
this.resHandler = resHandler;
this.jobGenerator = jobGenerator;
nextJobID = 0;
jobs = new HashMap<Integer, GridJob>();
reSendjobQueue = new LinkedList<GridJob>();
GridMaintainer maintainer = new GridMaintainer();
maintainer.start();
}
@Override
public void messageEvent(WorkerDataEvent e) {
try {
// ignores other messages than GridMessage
if(e.data instanceof GridMessage){
GridMessage msg = (GridMessage)e.data;
GridJob job = null;
switch(msg.messageType()){
case GridMessage.REGISTER:
e.network.send(e.socket, new GridMessage(GridMessage.INIT_DATA, 0, jobGenerator.initValues()));
break;
// Sending new data to compute to the client
case GridMessage.NEW_DATA:
if(!reSendjobQueue.isEmpty()){ // checks first if there is a job for recalculation
job = reSendjobQueue.poll();
job.renewTimeStamp();
}
else{ // generates new job
job = new GridJob(nextJobID,
jobGenerator.generateJob());
jobs.put(job.jobID, job);
nextJobID++;
}
GridMessage newMsg = new GridMessage(GridMessage.COMP_DATA, job.jobID, job.job);
e.network.send(e.socket, newMsg);
break;
// Received computation results
case GridMessage.COMP_SUCCESSFUL:
resHandler.resultEvent(msg.getJobQueueID(), true, msg.getData());
break;
case GridMessage.COMP_INCORRECT:
resHandler.resultEvent(msg.getJobQueueID(), false, msg.getData());
break;
case GridMessage.COMP_ERROR: // marks the job for recalculation
job = jobs.get(msg.getJobQueueID());
reSendjobQueue.add(job);
break;
}
}
} catch (IOException e1) {
e1.printStackTrace();
}
}
/**
* Changes the job timeout value
* @param min is the timeout in minutes
*/
public static void setJobTimeOut(int min){
JOB_TIMEOUT = 1000*60*min;
}
class GridMaintainer extends Thread{
/**
* Runs some behind the scenes stuff
* like job timeout.
*/
public void run(){
while(true){
long time = System.currentTimeMillis();
for(int jobID : jobs.keySet()){
if(time-jobs.get(jobID).timestamp > JOB_TIMEOUT){
reSendjobQueue.add(jobs.get(jobID));
}
}
try{Thread.sleep(1000*60*1);}catch(Exception e){};
}
}
}
}

38
src/zutil/net/nio/worker/grid/GridThread.java Normal file
View file

@ -0,0 +1,38 @@
package zutil.net.nio.worker.grid;
/**
* This interface is the thread that will do
* all the computation in the grid
*
* @author Ziver
*/
public abstract class GridThread implements Runnable{
/**
* The initial static and final data will be sent to this
* method.
*
* @param data is the static and or final data
*/
public abstract void setInitData(Object data);
public void run(){
while(true){
GridJob tmp = null;
try {
tmp = GridClient.getNextJob();
compute(tmp);
} catch (Exception e) {
e.printStackTrace();
if(tmp != null){
GridClient.jobError(tmp.jobID);
}
}
}
}
/**
* Compute the given data and return
* @param data
*/
public abstract void compute(GridJob data) throws Exception;
}