public void testDirectSockets() throws IOException, InterruptedException {
  final int DATA_LEN=200;
  final Random r=new Random();
  final byte[] data=new byte[DATA_LEN];
  r.nextBytes(data);
  final byte[] dst=new byte[DATA_LEN];
  int receiveBufferSize=-1;
  final InetSocketAddress serverAddr=new InetSocketAddress(getPort(0));
  final ServerSocket ss=new ServerSocket();
  try {
    assertTrue(ss.getChannel() == null);
    ss.bind(serverAddr);
    assertTrue(ss.getChannel() == null);
    receiveBufferSize=ss.getReceiveBufferSize();
    if (log.isInfoEnabled())     log.info("receiveBufferSize=" + receiveBufferSize + ", payloadSize="+ DATA_LEN);
    if (receiveBufferSize < DATA_LEN) {
      fail("Service socket receive buffer is smaller than test payload size: receiveBufferSize=" + receiveBufferSize + ", payloadSize="+ DATA_LEN);
    }
{
      InputStream instr=null;
      final SocketChannel cs1=SocketChannel.open();
      try {
        final boolean immediate1=cs1.connect(serverAddr);
        if (!immediate1) {
          if (!cs1.finishConnect()) {
            fail("Did not connect?");
          }
        }
        assertTrue(ss.getChannel() == null);
        final ByteBuffer src=ByteBuffer.wrap(data);
        cs1.write(src);
        final Socket readSckt1=accept(ss);
        instr=readSckt1.getInputStream();
        instr.read(dst);
        assertTrue(BytesUtil.bytesEqual(data,dst));
        assertTrue(ss.getChannel() == null);
        assertEquals(0,instr.available());
        cs1.write(ByteBuffer.wrap(data));
        cs1.close();
        assertTrue(readSckt1.isConnected());
        assertFalse(readSckt1.isClosed());
        try {
          cs1.write(ByteBuffer.wrap(data));
          fail("Expected closed channel exception");
        }
 catch (        ClosedChannelException e) {
        }
{
          final int rdlen=instr.read(dst);
          assertEquals(DATA_LEN,rdlen);
          assertTrue(BytesUtil.bytesEqual(data,dst));
        }
        assertEquals(-1,instr.read(dst));
        readSckt1.close();
        assertTrue(readSckt1.isClosed());
        assertEquals(-1,instr.read(dst));
        assertFalse(ss.isClosed());
        assertTrue(ss.getChannel() == null);
      }
  finally {
        cs1.close();
      }
      final int nrlen=instr.read(dst);
      assertEquals(-1,nrlen);
    }
    final SocketChannel cs2=SocketChannel.open();
    try {
      final boolean immediate2=cs2.connect(serverAddr);
      if (!immediate2) {
        if (!cs2.finishConnect()) {
          fail("Did not connect?");
        }
      }
      final Socket s2=accept(ss);
      final int wlen=cs2.write(ByteBuffer.wrap(data));
      assertEquals(DATA_LEN,wlen);
      final InputStream instr2=s2.getInputStream();
      instr2.read(dst);
      assertTrue(BytesUtil.bytesEqual(data,dst));
{
        instr2.close();
        assertTrue(cs2.isOpen());
        Thread.sleep(1000);
        assertTrue(cs2.isOpen());
        s2.close();
        assertTrue(cs2.isOpen());
        Thread.sleep(1000);
        assertTrue(cs2.isOpen());
      }
{
        final int writeSize=1;
        int nwritesOk=0;
        long nbytesReceived=0L;
        while (true) {
          try {
            final int wlen2=cs2.write(ByteBuffer.wrap(data,0,writeSize));
            assertEquals(writeSize,wlen2);
            nwritesOk++;
            nbytesReceived+=wlen2;
            assertTrue(cs2.isOpen());
            Thread.sleep(1000);
            assertTrue(cs2.isOpen());
          }
 catch (          IOException ex) {
            if (log.isInfoEnabled())             log.info("Expected exception: nwritesOk=" + nwritesOk + ", nbytesReceived="+ nbytesReceived+ ", ex="+ ex);
            break;
          }
        }
      }
      try {
        final Socket s3=accept(ss);
        fail("Expected timeout failure");
      }
 catch (      AssertionFailedError afe) {
      }
    }
  finally {
      cs2.close();
    }
  }
  finally {
    ss.close();
  }
}

public void testDirectSockets ( ) throws IOException , InterruptedException { final int DATA_LEN = 200 ; final Random r = new Random ( ) ; final byte [ ] data = new byte [ DATA_LEN ] ; r . nextBytes ( data ) ; final byte [ ] dst = new byte [ DATA_LEN ] ; int receiveBufferSize = - 1 ; final InetSocketAddress serverAddr = new InetSocketAddress ( getPort ( 0 ) ) ; final ServerSocket ss = new ServerSocket ( ) ; try { assertTrue ( ss . getChannel ( ) == null ) ; ss . bind ( serverAddr ) ; assertTrue ( ss . getChannel ( ) == null ) ; receiveBufferSize = ss . getReceiveBufferSize ( ) ; if ( log . isInfoEnabled ( ) ) log . info ( "receiveBufferSize=" + receiveBufferSize + ", payloadSize=" + DATA_LEN ) ; if ( receiveBufferSize < DATA_LEN ) { fail ( "Service socket receive buffer is smaller than test payload size: receiveBufferSize=" + receiveBufferSize + ", payloadSize=" + DATA_LEN ) ; } { InputStream instr = null ; final SocketChannel cs1 = SocketChannel . open ( ) ; try { final boolean immediate1 = cs1 . connect ( serverAddr ) ; if ( ! immediate1 ) { if ( ! cs1 . finishConnect ( ) ) { fail ( "Did not connect?" ) ; } } assertTrue ( ss . getChannel ( ) == null ) ; final ByteBuffer src = ByteBuffer . wrap ( data ) ; cs1 . write ( src ) ; final Socket readSckt1 = accept ( ss ) ; instr = readSckt1 . getInputStream ( ) ; instr . read ( dst ) ; assertTrue ( BytesUtil . bytesEqual ( data , dst ) ) ; assertTrue ( ss . getChannel ( ) == null ) ; assertEquals ( 0 , instr . available ( ) ) ; cs1 . write ( ByteBuffer . wrap ( data ) ) ; cs1 . close ( ) ; assertTrue ( readSckt1 . isConnected ( ) ) ; assertFalse ( readSckt1 . isClosed ( ) ) ; try { cs1 . write ( ByteBuffer . wrap ( data ) ) ; fail ( "Expected closed channel exception" ) ; } catch ( ClosedChannelException e ) { } { final int rdlen = instr . read ( dst ) ; assertEquals ( DATA_LEN , rdlen ) ; assertTrue ( BytesUtil . bytesEqual ( data , dst ) ) ; } assertEquals ( - 1 , instr . read ( dst ) ) ; readSckt1 . close ( ) ; assertTrue ( readSckt1 . isClosed ( ) ) ; assertEquals ( - 1 , instr . read ( dst ) ) ; assertFalse ( ss . isClosed ( ) ) ; assertTrue ( ss . getChannel ( ) == null ) ; } finally { cs1 . close ( ) ; } final int nrlen = instr . read ( dst ) ; assertEquals ( - 1 , nrlen ) ; } final SocketChannel cs2 = SocketChannel . open ( ) ; try { final boolean immediate2 = cs2 . connect ( serverAddr ) ; if ( ! immediate2 ) { if ( ! cs2 . finishConnect ( ) ) { fail ( "Did not connect?" ) ; } } final Socket s2 = accept ( ss ) ; final int wlen = cs2 . write ( ByteBuffer . wrap ( data ) ) ; assertEquals ( DATA_LEN , wlen ) ; final InputStream instr2 = s2 . getInputStream ( ) ; instr2 . read ( dst ) ; assertTrue ( BytesUtil . bytesEqual ( data , dst ) ) ; { instr2 . close ( ) ; assertTrue ( cs2 . isOpen ( ) ) ; Thread . sleep ( 1000 ) ; assertTrue ( cs2 . isOpen ( ) ) ; s2 . close ( ) ; assertTrue ( cs2 . isOpen ( ) ) ; Thread . sleep ( 1000 ) ; assertTrue ( cs2 . isOpen ( ) ) ; } { final int writeSize = 1 ; int nwritesOk = 0 ; long nbytesReceived = 0L ; while ( true ) { try { final int wlen2 = cs2 . write ( ByteBuffer . wrap ( data , 0 , writeSize ) ) ; assertEquals ( writeSize , wlen2 ) ; nwritesOk ++ ; nbytesReceived += wlen2 ; assertTrue ( cs2 . isOpen ( ) ) ; Thread . sleep ( 1000 ) ; assertTrue ( cs2 . isOpen ( ) ) ; } catch ( IOException ex ) { if ( log . isInfoEnabled ( ) ) log . info ( "Expected exception: nwritesOk=" + nwritesOk + ", nbytesReceived=" + nbytesReceived + ", ex=" + ex ) ; break ; } } } try { final Socket s3 = accept ( ss ) ; fail ( "Expected timeout failure" ) ; } catch ( AssertionFailedError afe ) { } } finally { cs2 . close ( ) ; } } finally { ss . close ( ) ; } }
the use of threaded tasks in the send / receive service makes it difficult to observer the socket state changes . so let ' s begin by writing some tests over the raw sockets . note that connecting and then immediately closing the socket is perfectly okay . . . . with an accept followed by a read ( ) of - 1 on the returned socket stream .
