Adjust the shutdown process to ensure that we can get messages generated during
the shutdown sequence out to our clients where desired. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4807 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
@@ -349,12 +349,6 @@ public class PresentsServer
|
|||||||
}
|
}
|
||||||
_downers = null;
|
_downers = null;
|
||||||
|
|
||||||
// shut down the connection manager (this will cease all network activity but not actually
|
|
||||||
// close the connections)
|
|
||||||
if (conmgr.isRunning()) {
|
|
||||||
conmgr.shutdown();
|
|
||||||
}
|
|
||||||
|
|
||||||
// shut down all shutdown participants
|
// shut down all shutdown participants
|
||||||
downers.apply(new ObserverList.ObserverOp<Shutdowner>() {
|
downers.apply(new ObserverList.ObserverOp<Shutdowner>() {
|
||||||
public boolean apply (Shutdowner downer) {
|
public boolean apply (Shutdowner downer) {
|
||||||
@@ -363,8 +357,13 @@ public class PresentsServer
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
// finally shut down the invoker and distributed object manager (The invoker does both for
|
// shut down the connection manager (this will cease all network activity but not actually
|
||||||
// us.)
|
// close the connections)
|
||||||
|
if (conmgr.isRunning()) {
|
||||||
|
conmgr.shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
// finally shut down the invoker and dobj manager (The invoker does both for us.)
|
||||||
invoker.shutdown();
|
invoker.shutdown();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -365,45 +365,8 @@ public class ConnectionManager extends LoopingThread
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// attempt to send any messages waiting on the overflow queues
|
// send any messages that are waiting on the outgoing overflow and message queues
|
||||||
if (_oflowqs.size() > 0) {
|
sendOutgoingMessages(iterStamp);
|
||||||
Iterator<OverflowQueue> oqiter = _oflowqs.values().iterator();
|
|
||||||
while (oqiter.hasNext()) {
|
|
||||||
OverflowQueue oq = oqiter.next();
|
|
||||||
try {
|
|
||||||
// try writing the messages in this overflow queue
|
|
||||||
if (oq.writeOverflowMessages(iterStamp)) {
|
|
||||||
// if they were all written, we can remove it
|
|
||||||
oqiter.remove();
|
|
||||||
}
|
|
||||||
|
|
||||||
} catch (IOException ioe) {
|
|
||||||
oq.conn.handleFailure(ioe);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// send any messages that are waiting on the outgoing queue
|
|
||||||
Tuple<Connection,byte[]> tup;
|
|
||||||
while ((tup = _outq.getNonBlocking()) != null) {
|
|
||||||
Connection conn = tup.left;
|
|
||||||
|
|
||||||
// if an overflow queue exists for this client, go ahead and slap the message on there
|
|
||||||
// because we can't send it until all other messages in their queue have gone out
|
|
||||||
OverflowQueue oqueue = _oflowqs.get(conn);
|
|
||||||
if (oqueue != null) {
|
|
||||||
int size = oqueue.size();
|
|
||||||
if ((size > 500) && (size % 50 == 0)) {
|
|
||||||
log.warning("Aiya, big overflow queue for " + conn + " [size=" + size +
|
|
||||||
", adding=" + tup.right + "].");
|
|
||||||
}
|
|
||||||
oqueue.add(tup.right);
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
// otherwise write the message out to the client directly
|
|
||||||
writeMessage(conn, tup.right, _oflowHandler);
|
|
||||||
}
|
|
||||||
|
|
||||||
// check for connections that have completed authentication
|
// check for connections that have completed authentication
|
||||||
AuthingConnection conn;
|
AuthingConnection conn;
|
||||||
@@ -517,6 +480,54 @@ public class ConnectionManager extends LoopingThread
|
|||||||
ready.clear();
|
ready.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Writes all queued overflow and normal messages to their respective sockets. Connections that
|
||||||
|
* already have established overflow queues will have their messages appended to their overflow
|
||||||
|
* queue instead so that they are delivered in the proper order.
|
||||||
|
*/
|
||||||
|
protected void sendOutgoingMessages (long iterStamp)
|
||||||
|
{
|
||||||
|
// first attempt to send any messages waiting on the overflow queues
|
||||||
|
if (_oflowqs.size() > 0) {
|
||||||
|
Iterator<OverflowQueue> oqiter = _oflowqs.values().iterator();
|
||||||
|
while (oqiter.hasNext()) {
|
||||||
|
OverflowQueue oq = oqiter.next();
|
||||||
|
try {
|
||||||
|
// try writing the messages in this overflow queue
|
||||||
|
if (oq.writeOverflowMessages(iterStamp)) {
|
||||||
|
// if they were all written, we can remove it
|
||||||
|
oqiter.remove();
|
||||||
|
}
|
||||||
|
|
||||||
|
} catch (IOException ioe) {
|
||||||
|
oq.conn.handleFailure(ioe);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// then send any new messages
|
||||||
|
Tuple<Connection,byte[]> tup;
|
||||||
|
while ((tup = _outq.getNonBlocking()) != null) {
|
||||||
|
Connection conn = tup.left;
|
||||||
|
|
||||||
|
// if an overflow queue exists for this client, go ahead and slap the message on there
|
||||||
|
// because we can't send it until all other messages in their queue have gone out
|
||||||
|
OverflowQueue oqueue = _oflowqs.get(conn);
|
||||||
|
if (oqueue != null) {
|
||||||
|
int size = oqueue.size();
|
||||||
|
if ((size > 500) && (size % 50 == 0)) {
|
||||||
|
log.warning("Aiya, big overflow queue for " + conn + " [size=" + size +
|
||||||
|
", adding=" + tup.right + "].");
|
||||||
|
}
|
||||||
|
oqueue.add(tup.right);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// otherwise write the message out to the client directly
|
||||||
|
writeMessage(conn, tup.right, _oflowHandler);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Writes a message out to a connection, passing the buck to the partial write handler if the
|
* Writes a message out to a connection, passing the buck to the partial write handler if the
|
||||||
* entire message could not be written.
|
* entire message could not be written.
|
||||||
@@ -603,6 +614,9 @@ public class ConnectionManager extends LoopingThread
|
|||||||
// documentation inherited
|
// documentation inherited
|
||||||
protected void didShutdown ()
|
protected void didShutdown ()
|
||||||
{
|
{
|
||||||
|
// take one last crack at the outgoing message queue
|
||||||
|
sendOutgoingMessages(System.currentTimeMillis());
|
||||||
|
|
||||||
// unbind our listening socket
|
// unbind our listening socket
|
||||||
try {
|
try {
|
||||||
_ssocket.socket().close();
|
_ssocket.socket().close();
|
||||||
|
|||||||
Reference in New Issue
Block a user