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

More code, more code, more kibbles and code.

Browse Source 
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2001-05-23 04:03:41 +00:00
parent d865c89bee
commit 6717f8d6e5
14 changed files with 402 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/client/Client.java
+3 -3
View File
@@ -1,5 +1,5 @@
//
// $Id: Client.java,v 1.1 2001/05/22 06:07:59 mdb Exp $
// $Id: Client.java,v 1.2 2001/05/23 04:03:40 mdb Exp $
package com.samskivert.cocktail.cher.client;
@@ -152,7 +152,7 @@ public class Client
case CLIENT_FAILED_TO_LOGON:
obs.clientFailedToLogon(this, cause);
break;
case CLIENT_CONNETION_FAILED:
case CLIENT_CONNECTION_FAILED:
obs.clientConnectionFailed(this, cause);
break;
case CLIENT_WILL_LOGOFF:
@@ -191,7 +191,7 @@ public class Client
// client observer codes
static final int CLIENT_DID_LOGON = 0;
static final int CLIENT_FAILED_TO_LOGON = 1;
static final int CLIENT_CONNETION_FAILED = 2;
static final int CLIENT_CONNECTION_FAILED = 2;
static final int CLIENT_WILL_LOGOFF = 3;
static final int CLIENT_DID_LOGOFF = 4;
}
src/java/com/threerings/presents/client/Communicator.java
+265 -22
View File
@@ -1,5 +1,5 @@
//
// $Id: Communicator.java,v 1.2 2001/05/22 21:51:29 mdb Exp $
// $Id: Communicator.java,v 1.3 2001/05/23 04:03:40 mdb Exp $
package com.samskivert.cocktail.cher.client;
@@ -10,8 +10,10 @@ import java.net.InetAddress;
import com.samskivert.util.Queue;
import com.samskivert.cocktail.cher.Log;
import com.samskivert.cocktail.cher.io.FramingOutputStream;
import com.samskivert.cocktail.cher.io.*;
import com.samskivert.cocktail.cher.io.ObjectStreamException;
import com.samskivert.cocktail.cher.net.*;
import com.samskivert.cocktail.cher.util.Codes;
/**
* The client performs all network I/O on separate threads (one for
@@ -69,12 +71,59 @@ class Communicator
* network connection. Also causes all communication threads to
* terminate.
*/
public void logoff ()
public synchronized void logoff ()
{
// if our socket is already closed, we've already taken care of
// this business
if (_socket == null) {
return;
}
// let our reader and writer know that it's time to go
if (_reader != null) {
// if logoff() is being called by the client as part of a
// normal shutdown, this will cause the reader thread to be
// interrupted and shutdown gracefully. if logoff is being
// called by the reader thread as a result of a failed socket,
// it won't interrupt itself as it is already shutting down
// gracefully. if the JVM is buggy and calling interrupt() on
// a thread that is blocked on a socket doesn't wake it up,
// then when we close() the socket a bit further down, we have
// another chance that the reader thread will wake up; this
// time slightly less gracefully because it will think there's
// a network error when in fact we're just shutting down, but
// at least it will cleanly exit
_reader.shutdown();
}
if (_writer != null) {
// shutting down the writer thread is simpler because we can
// post a termination message on the queue and be sure that it
// will receive it. we do run the risk that it is in the
// middle of trying to send a message when we close the
// socket, but in theory the send will fail, it will complain
// and then it will cleanly exit. if we were uber paranoid
// about JVMs misbehaving on simultaneous close()/write(), we
// could wait here for the writer thread to exit, but that
// makes me even more nervous because I know that some JVMs
// don't handle Thread.join() properly (hopefully no one is
// still using those JVMs but one can never be sure)
_writer.shutdown();
}
// close down our socket
try {
_socket.close();
} catch (IOException cle) {
Log.warning("Error closing failed socket: " + cle);
}
_socket = null;
// let the client observers know that we're logged off
_client.notifyObservers(Client.CLIENT_DID_LOGOFF, null);
}
/**
* Queues up the specified message for delivery to the server.
* Queues up the specified message for delivery upstream.
*/
public void postMessage (UpstreamMessage msg)
{
@@ -82,17 +131,46 @@ class Communicator
_msgq.append(msg);
}
protected void startWriter ()
/**
* Callback called by the reader when the authentication process
* completes successfully. Here we extract the bootstrap information
* for the client and start up the writer thread to manage the other
* half of our bi-directional message stream.
*/
protected synchronized void logonSucceeded (AuthResponseData data)
{
// extract bootstrap information
// create a new writer thread and start it up
if (_writer != null) {
throw new RuntimeException("Writer already started!?");
}
// create a new writer thread and start it up
_writer = new Writer();
_writer.start();
}
/**
* Callback called by the reader or writer thread when something goes
* awry with our socket connection to the server.
*/
protected synchronized void connectionFailed (IOException ioe)
{
// make sure the socket isn't already closed down (meaning we've
// already dealt with the failed connection)
if (_socket == null) {
return;
}
// let the client know that things went south
_client.notifyObservers(Client.CLIENT_CONNECTION_FAILED, ioe);
// and request that we go through the motions of logging off
logoff();
}
/**
* Callback called by the reader thread when it goes away.
*/
protected synchronized void readerDidExit ()
{
// clear out our reader reference
@@ -100,6 +178,19 @@ class Communicator
// let the client know when we finally go away
_client.communicatorDidExit();
Log.info("Reader thread exited.");
}
/**
* Callback called by the writer thread when it goes away.
*/
protected synchronized void writerDidExit ()
{
// clear out our writer reference
_writer = null;
Log.info("Writer thread exited.");
}
/**
@@ -109,7 +200,7 @@ class Communicator
throws IOException
{
// first we flatten the message so that we can measure it's length
msg.writeTo(_dout);
TypedObjectFactory.writeTo(_dout, msg);
// then write the framed message to actual output stream
_fout.writeFrameAndReset(_out);
}
@@ -121,11 +212,19 @@ class Communicator
protected DownstreamMessage receiveMessage ()
throws IOException
{
// read the frame size which comes first
int count = _din.readInt();
// read in the next message frame
_fin.clearAndReadFrame(_in);
// then use the typed object factory to read and decode the proper
// downstream message instance
return (DownstreamMessage)TypedObjectFactory.readFrom(_din);
}
// now read
return null;
/**
* Callback called by the reader thread when it has parsed a new
* message from the socket and wishes to have it processed.
*/
protected void processMessage (DownstreamMessage msg)
{
}
/**
@@ -154,10 +253,9 @@ class Communicator
return;
}
// once authenticated, we go into full-duplex mode,
// starting up another thread to listen for messages while
// we handle the delivery of messages
startWriter();
// now that we're authenticated, we manage the reading
// half of things by continuously reading messages from
// the socket and processing them
listen();
} finally {
@@ -166,6 +264,22 @@ class Communicator
}
}
/**
* 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();
}
}
protected void connect ()
throws IOException
{
@@ -180,13 +294,14 @@ class Communicator
// establish a socket connection to said server
_socket = new Socket(host, _client.getPort());
// create our input and output streams
InputStream in = _socket.getInputStream();
_din = new DataInputStream(new BufferedInputStream(in));
// get a handle on our input and output streams
_in = _socket.getInputStream();
_out = _socket.getOutputStream();
// we frame our messages here and then write them directly to
// the real output stream
// our messages are framed (preceded by their length), so we
// use these helper streams to manage the framing
_fin = new FramedInputStream();
_din = new DataInputStream(_fin);
_fout = new FramingOutputStream();
_dout = new DataOutputStream(_fout);
}
@@ -197,11 +312,74 @@ class Communicator
// construct an auth request and send it
AuthRequest req = new AuthRequest(_client.getCredentials());
sendMessage(req);
// now wait for the auth response
AuthResponse rsp = (AuthResponse)receiveMessage();
AuthResponseData data = rsp.getData();
// if the auth request failed, we want to let the communicator
// know by throwing a login exception
if (!data.code.equals(Codes.SUCCESS)) {
throw new LogonException(data.code);
}
// we're all clear. let the communicator know that we're in
logonSucceeded(data);
}
protected void listen ()
{
DownstreamMessage msg = null;
while (isRunning()) {
try {
// read the next message from the socket
msg = receiveMessage();
// process the message
processMessage(msg);
} catch (ObjectStreamException ose) {
Log.warning("Error decoding message: " + ose);
// move on to the next message
} catch (InterruptedIOException iioe) {
// somebody set up us the bomb! we've been interrupted
// which means that we're being shut down, so we
// simply fall through and isRunning() will return
// false next time through the loop
Log.info("Reader thread woken up in time to die.");
} catch (EOFException eofe) {
Log.info("Connection closed by peer.");
// nothing left for us to do
shutdown();
} catch (IOException ioe) {
// let the communicator know that our connection
// failed
connectionFailed(ioe);
// and shut ourselves down
shutdown();
} catch (Exception e) {
Log.warning("Error processing message [msg=" + msg +
", error=" + e + "].");
// move on to the next message
}
}
}
/**
* 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;
}
/**
@@ -213,6 +391,67 @@ class Communicator
{
public void run ()
{
try {
while (isRunning()) {
// 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) {
break;
}
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();
}
}
/**
* 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;
// post a bogus message to the outgoing queue to ensure that
// the writer thread notices that it's time to go
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. */
protected static class TerminationMessage extends UpstreamMessage
{
public short getType ()
{
return -1;
}
}
@@ -221,11 +460,15 @@ class Communicator
protected Writer _writer;
protected Socket _socket;
protected DataInputStream _din;
protected InputStream _in;
protected OutputStream _out;
protected Queue _msgq = new Queue();
/** We use this to frame our upstream messages. */
protected FramingOutputStream _fout;
protected DataOutputStream _dout;
/** We use this to frame our downstream messages. */
protected FramedInputStream _fin;
protected DataInputStream _din;
}