implement enough networking to play a game via relay (provided I don't

quit on either device.)  Uses java.nio, a single thread blocking on
select() and a single SocketChannel.  Outbound packets are queued, and
incoming packets reassembled then passed to the JNIThread to be sent
to the jni for parsing.
This commit is contained in:
eehouse 2010-01-31 21:26:21 +00:00
parent 15ce0a12da
commit caa5d6a55f
4 changed files with 228 additions and 8 deletions

View file

@ -170,6 +170,9 @@ public class BoardActivity extends Activity implements UtilCtxt {
m_view.startHandling( m_jniThread, m_jniGamePtr, m_gi );
m_jniThread.handle( JNICmd.CMD_START );
if ( null != m_xport ) {
m_xport.setReceiver( m_jniThread );
}
}
} // onCreate
@ -181,6 +184,10 @@ public class BoardActivity extends Activity implements UtilCtxt {
protected void onDestroy()
{
if ( null != m_xport ) {
m_xport.waitToStop();
}
if ( null != m_jniThread ) {
m_jniThread.waitToStop();
Utils.logf( "onDestroy(): waitToStop() returned" );
@ -341,7 +348,16 @@ public class BoardActivity extends Activity implements UtilCtxt {
}
m_timers[why] = new TimerRunnable( why, when, handle );
m_handler.postDelayed( m_timers[why], 500 );
int inHowLong;
switch ( why ) {
case UtilCtxt.TIMER_COMMS:
inHowLong = when * 1000;
break;
default:
inHowLong = 500;
}
m_handler.postDelayed( m_timers[why], inHowLong );
}
public void clearTimer( int why )

View file

@ -4,23 +4,202 @@ package org.eehouse.android.xw4;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
import java.nio.channels.ClosedChannelException;
import java.nio.ByteBuffer;
import java.net.InetSocketAddress;
import java.util.Vector;
import java.util.Iterator;
import junit.framework.Assert;
import org.eehouse.android.xw4.jni.*;
import org.eehouse.android.xw4.jni.JNIThread.*;
public class CommsTransport extends Thread implements TransportProcs {
private Selector m_selector;
private SocketChannel m_socketChannel;
private int m_jniGamePtr;
// private CommsAddrRec m_addr;
private boolean m_running = false;
private CommsAddrRec m_addr;
private JNIThread m_jniThread;
private boolean m_done = false;
private Vector<ByteBuffer> m_buffersOut;
private ByteBuffer m_bytesOut;
private ByteBuffer m_bytesIn;
// assembling inbound packet
private byte[] m_packetIn;
private int m_haveLen = -1;
public CommsTransport( int jniGamePtr )
{
m_jniGamePtr = jniGamePtr;
m_buffersOut = new Vector<ByteBuffer>();
m_bytesIn = ByteBuffer.allocate( 2048 );
}
public void setReceiver( JNIThread jnit )
{
m_jniThread = jnit;
}
public void waitToStop()
{
m_done = true;
m_selector.wakeup();
try {
join(100); // wait up to 1/10 second
} catch ( java.lang.InterruptedException ie ) {
Utils.logf( "got InterruptedException: " + ie.toString() );
}
}
@Override
public void run()
{
try {
m_selector = Selector.open();
m_socketChannel = SocketChannel.open();
m_socketChannel.configureBlocking( false );
InetSocketAddress isa
= new InetSocketAddress( m_addr.ip_relay_hostName,
m_addr.ip_relay_port );
m_socketChannel.connect( isa );
} catch ( java.io.IOException ioe ) {
Utils.logf( ioe.toString() );
}
while ( !m_done ) {
try {
int ops = figureOps();
Utils.logf( "calling with ops=" + ops );
m_socketChannel.register( m_selector, ops );
m_selector.select();
} catch ( ClosedChannelException cce ) {
Utils.logf( "exiting: " + cce.toString() );
break;
} catch ( java.io.IOException ioe ) {
Utils.logf( "exiting: " + ioe.toString() );
Utils.logf( ioe.toString() );
break;
}
Iterator iter = m_selector.selectedKeys().iterator();
while ( iter.hasNext() ) {
SelectionKey key = (SelectionKey)iter.next();
SocketChannel channel = (SocketChannel)key.channel();
iter.remove();
try {
if (key.isValid() && key.isConnectable()) {
Utils.logf( "socket is connectable" );
if ( !channel.finishConnect() ) {
key.cancel();
}
}
if (key.isValid() && key.isReadable()) {
m_bytesIn.clear(); // will wipe any pending data!
Utils.logf( "socket is readable; buffer has space for "
+ m_bytesIn.remaining() );
int nRead = channel.read( m_bytesIn );
if ( nRead == -1 ) {
channel.close();
} else {
addIncoming();
}
}
if (key.isValid() && key.isWritable()) {
Utils.logf( "socket is writable" );
getOut();
if ( null != m_bytesOut ) {
int nWritten = channel.write( m_bytesOut );
Utils.logf( "wrote " + nWritten + " bytes" );
}
}
} catch ( java.io.IOException ioe ) {
key.cancel();
}
}
}
}
private synchronized void putOut( final byte[] buf )
{
int len = buf.length;
ByteBuffer netbuf = ByteBuffer.allocate( len + 2 );
Utils.logf( "allocated outbound buffer of size " + len );
netbuf.putShort( (short)len );
netbuf.put( buf );
m_buffersOut.add( netbuf );
Assert.assertEquals( netbuf.remaining(), 0 );
if ( null != m_selector ) {
m_selector.wakeup(); // tell it it's got some writing to do
}
}
private synchronized void getOut()
{
if ( null != m_bytesOut && m_bytesOut.remaining() == 0 ) {
m_bytesOut = null;
}
if ( null == m_bytesOut && m_buffersOut.size() > 0 ) {
m_bytesOut = m_buffersOut.remove(0);
m_bytesOut.flip();
}
}
private synchronized int figureOps() {
int ops;
if ( m_socketChannel.isConnected() ) {
ops = SelectionKey.OP_READ;
if ( (null != m_bytesOut && m_bytesOut.hasRemaining())
|| m_buffersOut.size() > 0 ) {
ops |= SelectionKey.OP_WRITE;
}
} else {
ops = SelectionKey.OP_CONNECT;
}
return ops;
}
private void addIncoming( )
{
m_bytesIn.flip();
Utils.logf( "got " + m_bytesIn.remaining() + " bytes" );
for ( ; ; ) {
int len = m_bytesIn.remaining();
Utils.logf( "addIncoming(): remaining: " + len );
if ( len <= 0 ) {
break;
}
if ( null == m_packetIn ) { // we're not mid-packet
Assert.assertTrue( len > 1 ); // tell me if I see this case
if ( len == 1 ) { // half a length byte...
break; // can I leave it in the buffer?
} else {
m_packetIn = new byte[m_bytesIn.getShort()];
m_haveLen = 0;
}
} else { // we're mid-packet
int wantLen = m_packetIn.length - m_haveLen;
if ( wantLen > len ) {
wantLen = len;
}
m_bytesIn.get( m_packetIn, m_haveLen, wantLen );
m_haveLen += wantLen;
if ( m_haveLen == m_packetIn.length ) {
// send completed packet
Utils.logf( "got full packet!!" );
m_jniThread.handle( JNICmd.CMD_RECEIVE, m_packetIn );
m_packetIn = null;
}
}
}
}
// TransportProcs interface
@ -28,14 +207,27 @@ public class CommsTransport extends Thread implements TransportProcs {
{
Utils.logf( "CommsTransport::transportSend" );
CommsAddrRec addr = faddr;
if ( null == addr ) {
addr = new CommsAddrRec();
XwJNI.comms_getAddr( m_jniGamePtr, addr );
if ( null == m_addr ) {
if ( null == faddr ) {
m_addr = new CommsAddrRec();
XwJNI.comms_getAddr( m_jniGamePtr, m_addr );
} else {
m_addr = new CommsAddrRec( faddr );
}
}
// add this packet to queue
putOut( buf );
// start the read/write thread. Note: server needs to start
// before first asked to send. For relay, though, that'll
// happen. Not sure about other transport e.g. BT where
// server needs to get into listening mode.
if ( !m_running ) {
m_running = true;
start();
}
Utils.logf( "CommsTransport::transportSend(" + addr.ip_relay_hostName +
") called!!!" );
return buf.length;
}

View file

@ -19,6 +19,7 @@ public class JNIThread extends Thread {
CMD_LAYOUT,
CMD_START,
CMD_DO,
CMD_RECEIVE,
CMD_PEN_DOWN,
CMD_PEN_MOVE,
CMD_PEN_UP,
@ -189,6 +190,12 @@ public class JNIThread extends Thread {
draw = XwJNI.server_do( m_jniGamePtr );
break;
case CMD_RECEIVE:
draw = XwJNI.game_receiveMessage( m_jniGamePtr,
(byte[])args[0] );
handle( JNICmd.CMD_DO );
break;
case CMD_PEN_DOWN:
draw = XwJNI.board_handlePenDown( m_jniGamePtr,
((Integer)args[0]).intValue(),

View file

@ -16,6 +16,11 @@ public interface UtilCtxt {
int playerNum, String[] texts );
boolean engineProgressCallback();
// Values for why; should be enums
public static final int TIMER_PENDOWN = 1;
public static final int TIMER_TIMERTICK = 2;
public static final int TIMER_COMMS = 3;
public static final int TIMER_SLOWROBOT = 4;
void setTimer( int why, int when, int handle );
void clearTimer( int why );
void requestTime();