This commit is contained in:
parent
16b9f1f265
commit
c0bf13e493
1 changed files with 69 additions and 57 deletions
|
|
@ -118,11 +118,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
* @param data The data to send
|
* @param data The data to send
|
||||||
*/
|
*/
|
||||||
public void send(SocketChannel socket, byte[] data) {
|
public void send(SocketChannel socket, byte[] data) {
|
||||||
synchronized (pendingChanges) {
|
queueSend(socket,data);
|
||||||
// Indicate we want the interest ops set changed
|
|
||||||
pendingChanges.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
|
|
||||||
queueSend(socket,data);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -133,6 +129,11 @@ public abstract class NioNetwork implements Runnable {
|
||||||
*/
|
*/
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
protected void queueSend(SocketChannel socket, byte[] data){
|
protected void queueSend(SocketChannel socket, byte[] data){
|
||||||
|
synchronized (pendingChanges) {
|
||||||
|
// Indicate we want the interest ops set changed
|
||||||
|
pendingChanges.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
|
||||||
|
}
|
||||||
|
System.out.println("pendingData");
|
||||||
// And queue the data we want written
|
// And queue the data we want written
|
||||||
synchronized (pendingData) {
|
synchronized (pendingData) {
|
||||||
List<ByteBuffer> queue = pendingData.get(socket);
|
List<ByteBuffer> queue = pendingData.get(socket);
|
||||||
|
|
@ -143,6 +144,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
queue.add(ByteBuffer.wrap(data));
|
queue.add(ByteBuffer.wrap(data));
|
||||||
|
|
||||||
}
|
}
|
||||||
|
System.out.println("selector.wakeup();");
|
||||||
// Finally, wake up our selecting thread so it can make the required changes
|
// Finally, wake up our selecting thread so it can make the required changes
|
||||||
selector.wakeup();
|
selector.wakeup();
|
||||||
}
|
}
|
||||||
|
|
@ -159,6 +161,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
case ChangeRequest.CHANGEOPS:
|
case ChangeRequest.CHANGEOPS:
|
||||||
SelectionKey key = change.socket.keyFor(selector);
|
SelectionKey key = change.socket.keyFor(selector);
|
||||||
key.interestOps(change.ops);
|
key.interestOps(change.ops);
|
||||||
|
System.out.println("change.ops "+change.ops);
|
||||||
break;
|
break;
|
||||||
case ChangeRequest.REGISTER:
|
case ChangeRequest.REGISTER:
|
||||||
change.socket.register(selector, change.ops);
|
change.socket.register(selector, change.ops);
|
||||||
|
|
@ -167,7 +170,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
}
|
}
|
||||||
pendingChanges.clear();
|
pendingChanges.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Wait for an event one of the registered channels
|
// Wait for an event one of the registered channels
|
||||||
selector.select();
|
selector.select();
|
||||||
|
|
||||||
|
|
@ -181,19 +184,21 @@ public abstract class NioNetwork implements Runnable {
|
||||||
// Check what event is available and deal with it
|
// Check what event is available and deal with it
|
||||||
if (key.isAcceptable()) {
|
if (key.isAcceptable()) {
|
||||||
accept(key);
|
accept(key);
|
||||||
//System.out.println("Accepting new Connection!! connection count:"+selector.keys().size());
|
System.out.println("Accepting new Connection!! connection count:"+selector.keys().size());
|
||||||
}
|
}
|
||||||
else if (key.isConnectable()) {
|
else if (key.isConnectable()) {
|
||||||
|
//key.interestOps(SelectionKey.OP_READ);
|
||||||
closeConnection(key);
|
closeConnection(key);
|
||||||
//System.out.println("Disconnecting Connection!! connection count:"+selector.keys().size());
|
System.out.println("Disconnecting Connection!! connection count:"+selector.keys().size());
|
||||||
}
|
}
|
||||||
else if (key.isReadable()) {
|
else if (key.isReadable()) {
|
||||||
read(key);
|
read(key);
|
||||||
//System.out.println("Reading");
|
System.out.println("Reading");
|
||||||
}
|
}
|
||||||
else if (key.isWritable()) {
|
else if (key.isWritable()) {
|
||||||
|
System.out.println("Writing");
|
||||||
write(key);
|
write(key);
|
||||||
//System.out.println("Writing");
|
System.out.println("Writing");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -223,6 +228,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
InetSocketAddress remoteAdr = (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
|
InetSocketAddress remoteAdr = (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
|
||||||
if(!clients.containsValue(remoteAdr)){
|
if(!clients.containsValue(remoteAdr)){
|
||||||
clients.put(remoteAdr, new ClientData(socketChannel));
|
clients.put(remoteAdr, new ClientData(socketChannel));
|
||||||
|
System.out.println("New Connection("+remoteAdr+")!!! Count: "+clients.size());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -231,7 +237,8 @@ public abstract class NioNetwork implements Runnable {
|
||||||
*/
|
*/
|
||||||
private void read(SelectionKey key) throws IOException {
|
private void read(SelectionKey key) throws IOException {
|
||||||
SocketChannel socketChannel = (SocketChannel) key.channel();
|
SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||||
|
InetSocketAddress remoteAdr = (InetSocketAddress) socketChannel.socket().getRemoteSocketAddress();
|
||||||
|
|
||||||
// Clear out our read buffer so it's ready for new data
|
// Clear out our read buffer so it's ready for new data
|
||||||
readBuffer.clear();
|
readBuffer.clear();
|
||||||
|
|
||||||
|
|
@ -244,6 +251,8 @@ public abstract class NioNetwork implements Runnable {
|
||||||
// the selection key and close the channel.
|
// the selection key and close the channel.
|
||||||
key.cancel();
|
key.cancel();
|
||||||
socketChannel.close();
|
socketChannel.close();
|
||||||
|
clients.remove(remoteAdr);
|
||||||
|
System.out.println("Connection Forced Closed("+remoteAdr+")!!! Count: "+clients.size());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -252,6 +261,8 @@ public abstract class NioNetwork implements Runnable {
|
||||||
// same from our end and cancel the channel.
|
// same from our end and cancel the channel.
|
||||||
key.channel().close();
|
key.channel().close();
|
||||||
key.cancel();
|
key.cancel();
|
||||||
|
clients.remove(remoteAdr);
|
||||||
|
System.out.println("Connection Closed("+remoteAdr+")!!! Count: "+clients.size());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -259,7 +270,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
// to the client
|
// to the client
|
||||||
byte[] rspByteData = new byte[numRead];
|
byte[] rspByteData = new byte[numRead];
|
||||||
System.arraycopy(readBuffer.array(), 0, rspByteData, 0, numRead);
|
System.arraycopy(readBuffer.array(), 0, rspByteData, 0, numRead);
|
||||||
|
|
||||||
Object rspData = Converter.toObject(rspByteData);
|
Object rspData = Converter.toObject(rspByteData);
|
||||||
|
|
||||||
if(rspData instanceof SystemMessage){
|
if(rspData instanceof SystemMessage){
|
||||||
|
|
@ -284,51 +295,7 @@ public abstract class NioNetwork implements Runnable {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Initieates a socket to the server
|
|
||||||
*/
|
|
||||||
protected SocketChannel initiateConnection(InetSocketAddress address) throws IOException {
|
|
||||||
// Create a non-blocking socket channel
|
|
||||||
SocketChannel socketChannel = SocketChannel.open();
|
|
||||||
socketChannel.configureBlocking(false);
|
|
||||||
|
|
||||||
// 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 And Server ResponseEvent
|
|
||||||
*/
|
|
||||||
private void handleResponse(SocketChannel socketChannel, Object rspData) throws IOException {
|
|
||||||
|
|
||||||
// Look up the handler for this channel
|
|
||||||
ResponseEvent handler = (ResponseEvent) rspEvents.get(socketChannel);
|
|
||||||
|
|
||||||
// And pass the response to it
|
|
||||||
if (handler.handleResponse(rspData)) {
|
|
||||||
// The handler has seen enough, close the connection
|
|
||||||
socketChannel.close();
|
|
||||||
socketChannel.keyFor(selector).cancel();
|
|
||||||
}
|
|
||||||
|
|
||||||
rspEvents.remove(socketChannel);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Client and Server
|
* Client and Server
|
||||||
*/
|
*/
|
||||||
|
|
@ -359,10 +326,55 @@ public abstract class NioNetwork implements Runnable {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Client And Server ResponseEvent
|
||||||
|
*/
|
||||||
|
private void handleResponse(SocketChannel socketChannel, Object rspData) throws IOException {
|
||||||
|
|
||||||
|
// Look up the handler for this channel
|
||||||
|
ResponseEvent handler = (ResponseEvent) rspEvents.get(socketChannel);
|
||||||
|
|
||||||
|
// And pass the response to it
|
||||||
|
if (handler.handleResponse(rspData)) {
|
||||||
|
// The handler has seen enough, close the connection
|
||||||
|
//socketChannel.close();
|
||||||
|
//socketChannel.keyFor(selector).cancel();
|
||||||
|
}
|
||||||
|
|
||||||
|
rspEvents.remove(socketChannel);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Initieates a socket to the server
|
||||||
|
*/
|
||||||
|
protected SocketChannel initiateConnection(InetSocketAddress address) throws IOException {
|
||||||
|
// Create a non-blocking socket channel
|
||||||
|
SocketChannel socketChannel = SocketChannel.open();
|
||||||
|
socketChannel.configureBlocking(false);
|
||||||
|
System.out.println(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
|
* Client
|
||||||
*/
|
*/
|
||||||
private void closeConnection(SelectionKey key) throws IOException {
|
protected void closeConnection(SelectionKey key) throws IOException {
|
||||||
SocketChannel socketChannel = (SocketChannel) key.channel();
|
SocketChannel socketChannel = (SocketChannel) key.channel();
|
||||||
|
|
||||||
// Finish the connection. If the connection operation failed
|
// Finish the connection. If the connection operation failed
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue