AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use LoopingThread for Reader and Writer.

Browse Source 
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@11 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2001-05-30 23:51:39 +00:00
parent f6be4c148d
commit bd58cbd199
1 changed files with 82 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/client/Communicator.java
+82 -125
View File
@@ -1,5 +1,5 @@
// //
// $Id: Communicator.java,v 1.4 2001/05/29 03:27:59 mdb Exp $ // $Id: Communicator.java,v 1.5 2001/05/30 23:51:39 mdb Exp $
package com.samskivert.cocktail.cher.client; package com.samskivert.cocktail.cher.client;
@@ -7,6 +7,7 @@ import java.io.*;
import java.net.Socket; import java.net.Socket;
import java.net.InetAddress; import java.net.InetAddress;
import com.samskivert.util.LoopingThread;
import com.samskivert.util.Queue; import com.samskivert.util.Queue;
import com.samskivert.cocktail.cher.Log; import com.samskivert.cocktail.cher.Log;
@@ -38,7 +39,7 @@ import com.samskivert.cocktail.cher.util.Codes;
* - write loop * - write loop
* </pre> * </pre>
*/ */
class Communicator public class Communicator
{ {
/** /**
* Creates a new communicator instance which is associated with the * Creates a new communicator instance which is associated with the
@@ -246,52 +247,25 @@ class Communicator
* things, but the general flow of the reader thread is encapsulated * things, but the general flow of the reader thread is encapsulated
* in this class. * in this class.
*/ */
protected class Reader extends Thread protected class Reader extends LoopingThread
{ {
public void run () protected void willStart ()
{ {
// first we connect and authenticate with the server
try { try {
// first we connect and authenticate with the server // connect to the server
try { connect();
// connect to the server
connect();
// then authenticate // then authenticate
logon(); logon();
} catch (Exception e) { } catch (Exception e) {
Log.info("Logon failed: " + e); Log.info("Logon failed: " + e);
Log.logStackTrace(e); Log.logStackTrace(e);
// let the observers know that we've failed // let the observers know that we've failed
_client.notifyObservers(Client.CLIENT_FAILED_TO_LOGON, e); _client.notifyObservers(Client.CLIENT_FAILED_TO_LOGON, e);
// and terminate our communicator thread // and terminate our communicator thread
return; shutdown();
}
// now that we're authenticated, we manage the reading
// half of things by continuously reading messages from
// the socket and processing them
listen();
} finally {
// let the communicator know when we finally go away
readerDidExit();
}
}
/**
* Informs the reader thread that it's no longer running. The next
* time through the read loop, it will exit.
*/
public synchronized void shutdown ()
{
_running = false;
// if we are not the reader thread, then we want to interrupt
// the reader thread as it may be blocked listening to the
// socket
if (Thread.currentThread() != this) {
interrupt();
} }
} }
@@ -329,8 +303,10 @@ class Communicator
sendMessage(req); sendMessage(req);
// now wait for the auth response // now wait for the auth response
Log.info("Waiting for auth response.");
AuthResponse rsp = (AuthResponse)receiveMessage(); AuthResponse rsp = (AuthResponse)receiveMessage();
AuthResponseData data = rsp.getData(); AuthResponseData data = rsp.getData();
Log.info("Got auth response: " + data);
// if the auth request failed, we want to let the communicator // if the auth request failed, we want to let the communicator
// know by throwing a login exception // know by throwing a login exception
@@ -342,59 +318,59 @@ class Communicator
logonSucceeded(data); logonSucceeded(data);
} }
protected void listen () // now that we're authenticated, we manage the reading
// half of things by continuously reading messages from
// the socket and processing them
protected void iterate ()
{ {
DownstreamMessage msg = null; DownstreamMessage msg = null;
while (isRunning()) { try {
try { // read the next message from the socket
// read the next message from the socket msg = receiveMessage();
msg = receiveMessage();
// process the message // process the message
processMessage(msg); processMessage(msg);
} catch (ObjectStreamException ose) { } catch (ObjectStreamException ose) {
Log.warning("Error decoding message: " + ose); Log.warning("Error decoding message: " + ose);
// move on to the next message
} catch (InterruptedIOException iioe) { } catch (InterruptedIOException iioe) {
// somebody set up us the bomb! we've been interrupted // somebody set up us the bomb! we've been interrupted
// which means that we're being shut down, so we // which means that we're being shut down, so we just
// simply fall through and isRunning() will return // report it and return from iterate() like a good monkey
// false next time through the loop Log.info("Reader thread woken up in time to die.");
Log.info("Reader thread woken up in time to die.");
} catch (EOFException eofe) { } catch (EOFException eofe) {
Log.info("Connection closed by peer."); Log.info("Connection closed by peer.");
// nothing left for us to do // nothing left for us to do
shutdown(); shutdown();
} catch (IOException ioe) { } catch (IOException ioe) {
// let the communicator know that our connection // let the communicator know that our connection failed
// failed connectionFailed(ioe);
connectionFailed(ioe); // and shut ourselves down
// and shut ourselves down shutdown();
shutdown();
} catch (Exception e) { } catch (Exception e) {
Log.warning("Error processing message [msg=" + msg + Log.warning("Error processing message [msg=" + msg +
", error=" + e + "]."); ", error=" + e + "].");
// move on to the next message
}
} }
} }
/** protected void didShutdown ()
* Must access _running via this member function to ensure that we
* are Chapter 17 compliant.
*/
protected synchronized boolean isRunning ()
{ {
return _running; // let the communicator know when we finally go away
readerDidExit();
} }
protected boolean _running = true; protected void kick ()
{
// we want to interrupt the reader thread as it may be blocked
// listening to the socket; this is only called if the reader
// thread doesn't shut itself down
interrupt();
}
} }
/** /**
@@ -402,63 +378,44 @@ class Communicator
* to the <code>Communicator</code> class to do things, but the * to the <code>Communicator</code> class to do things, but the
* general flow of the writer thread is encapsulated in this class. * general flow of the writer thread is encapsulated in this class.
*/ */
protected class Writer extends Thread protected class Writer extends LoopingThread
{ {
public void run () protected void iterate ()
{ {
// fetch the next message from the queue
UpstreamMessage msg = (UpstreamMessage)_msgq.get();
// if this is a termination message, we're being
// requested to exit, so we want to bail now rather
// than continuing
if (msg instanceof TerminationMessage) {
return;
}
try { try {
while (isRunning()) { // write the message out the socket
// fetch the next message from the queue sendMessage(msg);
UpstreamMessage msg = (UpstreamMessage)_msgq.get();
// if this is a termination message, we're being } catch (IOException ioe) {
// requested to exit, so we want to bail now rather // let the communicator know if we have any
// than continuing // problems
if (msg instanceof TerminationMessage) { connectionFailed(ioe);
break; // and bail
} shutdown();
try {
// write the message out the socket
sendMessage(msg);
} catch (IOException ioe) {
// let the communicator know if we have any
// problems
connectionFailed(ioe);
// and bail
shutdown();
}
}
} finally {
writerDidExit();
} }
} }
/** protected void didShutdown ()
* Informs the writer thread that it's no longer running. The next
* time through the write loop, it will exit.
*/
public synchronized void shutdown ()
{ {
_running = false; writerDidExit();
}
protected void kick ()
{
// post a bogus message to the outgoing queue to ensure that // post a bogus message to the outgoing queue to ensure that
// the writer thread notices that it's time to go // the writer thread notices that it's time to go
postMessage(new TerminationMessage()); postMessage(new TerminationMessage());
} }
/**
* Must access _running via this member function to ensure that we
* are Chapter 17 compliant.
*/
protected synchronized boolean isRunning ()
{
return _running;
}
protected boolean _running = true;
} }
/** This is used to terminate the writer thread. */ /** This is used to terminate the writer thread. */