ActiveIO - 1.0

Clover coverage report - ActiveIO - 1.0

Coverage timestamp: Fri Apr 22 2005 14:27:22 PDT

FRAMES NO FRAMES

file stats:	LOC:	182		Methods:	10
	NCLOC:	125		Classes:	1

30 day Evaluation Version distributed via the Maven Jar Repository. Clover is not free. You have 30 days to evaluate it. Please visit http://www.thecortex.net/clover to obtain a licensed version of Clover

Source file

Conditionals

Statements

Methods

TOTAL

NIOAsynchChannel.java

57.1%

67.7%

70%

65%

 /** 
  * 
  * Copyright 2004 Hiram Chirino
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); 
  * you may not use this file except in compliance with the License. 
  * You may obtain a copy of the License at 
  * 
  * http://www.apache.org/licenses/LICENSE-2.0
  * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS, 
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
  * See the License for the specific language governing permissions and 
  * limitations under the License. 
  * 
  **/
 
 package org.activeio.net;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 
 import org.activeio.AsynchChannel;
 import org.activeio.AsynchChannelListener;
 import org.activeio.Packet;
 import org.activeio.Packet.ByteSequence;
 import org.activeio.net.NIOAsynchChannelSelectorManager.SelectorManagerListener;
 import org.activeio.net.NIOAsynchChannelSelectorManager.SocketChannelAsynchChannelSelection;
 import org.activeio.packet.ByteBufferPacket;
 import org.activeio.packet.EOSPacket;
 
 /**
  * @version $Revision$
  */
 final public class NIOAsynchChannel extends NIOBaseChannel implements AsynchChannel {
 
     private AsynchChannelListener channelListener;
     private SocketChannelAsynchChannelSelection selection;
     private ByteBuffer inputByteBuffer;
     private boolean running;
 
     public NIOAsynchChannel(SocketChannel socketChannel, boolean useDirect) throws IOException {
         super(socketChannel, useDirect);
 
         socketChannel.configureBlocking(false);                
         selection = NIOAsynchChannelSelectorManager.register(socketChannel, new SelectorManagerListener(){
             public void onSelect(SocketChannelAsynchChannelSelection selection) {
                 String origName = Thread.currentThread().getName();
                 if (selection.isReadable())
                 try {
                     Thread.currentThread().setName(NIOAsynchChannel.this.toString());
                     serviceRead();
                  } catch ( Throwable e ) {
                      System.err.println("ActiveIO unexpected error: ");
                      e.printStackTrace(System.err);
                  } finally {
                      Thread.currentThread().setName(origName);
                  }
             }
         });
         
     }
     
     private void serviceRead() {
         try {
             
             while( true ) {
                 
                 if( inputByteBuffer==null || !inputByteBuffer.hasRemaining() ) {
                     inputByteBuffer = allocateBuffer();
                 }
     
                 int size = socketChannel.read(inputByteBuffer);
                 if( size == -1 ) {
                     this.channelListener.onPacket( EOSPacket.EOS_PACKET );
                     selection.close();
                     break;
                 }
 
                 if( size==0 ) {
                     break;
                 }
                 
                 // Per Mike Spile, some plaforms read 1 byte of data on the first read, and then
                 // a but load of data on the second read.  Try to load the butload here
                 if( size == 1 && inputByteBuffer.hasRemaining() ) {
                     int size2 = socketChannel.read(inputByteBuffer);
                     if( size2 > 0 )
                             size += size2;
                 }
                 
                 ByteBuffer remaining = inputByteBuffer.slice();            
                 Packet data = new ByteBufferPacket(((ByteBuffer)inputByteBuffer.flip()).slice());
                 this.channelListener.onPacket( data );
                             
                 // Keep the remaining buffer around to fill with data.
                 inputByteBuffer = remaining;
                 
                 if( inputByteBuffer.hasRemaining() )
                     break;
             }
             
         } catch (IOException e) {
             this.channelListener.onPacketError(e);
         }
     }
     
     synchronized public void write(Packet packet) throws IOException {
         
         ByteBuffer data;
         if( packet.getClass()==ByteBufferPacket.class ) {
             data = ((ByteBufferPacket)packet).getByteBuffer();            
         } else {
             ByteSequence sequence = packet.asByteSequence();
             data = ByteBuffer.wrap(sequence.getData(), sequence.getOffset(), sequence.getLength());
         }
 
         long delay=1;
         while( data.hasRemaining() ) {
             
             // Since the write is non-blocking, all the data may not have been written.
             int r1 = data.remaining();        
             socketChannel.write( data );        
             int r2 = data.remaining();
             
             // We may need to do a little bit of sleeping to avoid a busy loop.
             // Slow down if no data was written out.. 
             if( r2>0 && r1-r2==0 ) {
                 try {
                     // Use exponential rollback to increase sleep time.
                     Thread.sleep(delay);
                     delay *= 5;
                     if( delay > 1000*1 ) {
                         delay = 1000;
                     }
                 } catch (InterruptedException e) {
                     throw new InterruptedIOException();
                 }
             } else {
                 delay = 1;
             }
         }
     }
 
     public void flush() throws IOException {
     }
 
     public void setAsynchChannelListener(AsynchChannelListener channelListener) {
         this.channelListener = channelListener;
     }
 
     public AsynchChannelListener getAsynchChannelListener() {
         return channelListener;
     }
 
     public void dispose() {
         if( running && channelListener!=null ) {
             channelListener.onPacketError(new SocketException("Socket closed."));
         }
         selection.close();
         super.dispose();
     }
 
     public void start() throws IOException {
         if( running )
             return;
         running=true;
         selection.setInterestOps(SelectionKey.OP_READ);
     }
 
     public void stop(long timeout) throws IOException {
         if( !running )
             return;
         running=false;
         selection.setInterestOps(0);        
     }
  }

1		/**
2		*
3		* Copyright 2004 Hiram Chirino
4		*
5		* Licensed under the Apache License, Version 2.0 (the "License");
6		* you may not use this file except in compliance with the License.
7		* You may obtain a copy of the License at
8		*
9		* http://www.apache.org/licenses/LICENSE-2.0
10		*
11		* Unless required by applicable law or agreed to in writing, software
12		* distributed under the License is distributed on an "AS IS" BASIS,
13		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14		* See the License for the specific language governing permissions and
15		* limitations under the License.
16		*
17		**/
18
19		package org.activeio.net;
20
21		import java.io.IOException;
22		import java.io.InterruptedIOException;
23		import java.net.SocketException;
24		import java.nio.ByteBuffer;
25		import java.nio.channels.SelectionKey;
26		import java.nio.channels.SocketChannel;
27
28		import org.activeio.AsynchChannel;
29		import org.activeio.AsynchChannelListener;
30		import org.activeio.Packet;
31		import org.activeio.Packet.ByteSequence;
32		import org.activeio.net.NIOAsynchChannelSelectorManager.SelectorManagerListener;
33		import org.activeio.net.NIOAsynchChannelSelectorManager.SocketChannelAsynchChannelSelection;
34		import org.activeio.packet.ByteBufferPacket;
35		import org.activeio.packet.EOSPacket;
36
37		/**
38		* @version $Revision$
39		*/
40		final public class NIOAsynchChannel extends NIOBaseChannel implements AsynchChannel {
41
42		private AsynchChannelListener channelListener;
43		private SocketChannelAsynchChannelSelection selection;
44		private ByteBuffer inputByteBuffer;
45		private boolean running;
46
47	15	public NIOAsynchChannel(SocketChannel socketChannel, boolean useDirect) throws IOException {
48	15	super(socketChannel, useDirect);
49
50	16	socketChannel.configureBlocking(false);
51	16	selection = NIOAsynchChannelSelectorManager.register(socketChannel, new SelectorManagerListener(){
52	2006	public void onSelect(SocketChannelAsynchChannelSelection selection) {
53	2006	String origName = Thread.currentThread().getName();
54	2006	if (selection.isReadable())
55	2006	try {
56	2006	Thread.currentThread().setName(NIOAsynchChannel.this.toString());
57	2006	serviceRead();
58		} catch ( Throwable e ) {
59	0	System.err.println("ActiveIO unexpected error: ");
60	0	e.printStackTrace(System.err);
61		} finally {
62	2006	Thread.currentThread().setName(origName);
63		}
64		}
65		});
66
67		}
68
69	2006	private void serviceRead() {
70	2006	try {
71
72	2006	while( true ) {
73
74	2008	if( inputByteBuffer==null \|\| !inputByteBuffer.hasRemaining() ) {
75	8	inputByteBuffer = allocateBuffer();
76		}
77
78	2008	int size = socketChannel.read(inputByteBuffer);
79	2008	if( size == -1 ) {
80	2	this.channelListener.onPacket( EOSPacket.EOS_PACKET );
81	2	selection.close();
82	2	break;
83		}
84
85	2006	if( size==0 ) {
86	0	break;
87		}
88
89		// Per Mike Spile, some plaforms read 1 byte of data on the first read, and then
90		// a but load of data on the second read. Try to load the butload here
91	2006	if( size == 1 && inputByteBuffer.hasRemaining() ) {
92	0	int size2 = socketChannel.read(inputByteBuffer);
93	0	if( size2 > 0 )
94	0	size += size2;
95		}
96
97	2006	ByteBuffer remaining = inputByteBuffer.slice();
98	2006	Packet data = new ByteBufferPacket(((ByteBuffer)inputByteBuffer.flip()).slice());
99	2006	this.channelListener.onPacket( data );
100
101		// Keep the remaining buffer around to fill with data.
102	2006	inputByteBuffer = remaining;
103
104	2006	if( inputByteBuffer.hasRemaining() )
105	2004	break;
106		}
107
108		} catch (IOException e) {
109	0	this.channelListener.onPacketError(e);
110		}
111		}
112
113	2004	synchronized public void write(Packet packet) throws IOException {
114
115	2004	ByteBuffer data;
116	2004	if( packet.getClass()==ByteBufferPacket.class ) {
117	2004	data = ((ByteBufferPacket)packet).getByteBuffer();
118		} else {
119	0	ByteSequence sequence = packet.asByteSequence();
120	0	data = ByteBuffer.wrap(sequence.getData(), sequence.getOffset(), sequence.getLength());
121		}
122
123	2004	long delay=1;
124	2004	while( data.hasRemaining() ) {
125
126		// Since the write is non-blocking, all the data may not have been written.
127	2004	int r1 = data.remaining();
128	2004	socketChannel.write( data );
129	2004	int r2 = data.remaining();
130
131		// We may need to do a little bit of sleeping to avoid a busy loop.
132		// Slow down if no data was written out..
133	2004	if( r2>0 && r1-r2==0 ) {
134	0	try {
135		// Use exponential rollback to increase sleep time.
136	0	Thread.sleep(delay);
137	0	delay *= 5;
138	0	if( delay > 1000*1 ) {
139	0	delay = 1000;
140		}
141		} catch (InterruptedException e) {
142	0	throw new InterruptedIOException();
143		}
144		} else {
145	2004	delay = 1;
146		}
147		}
148		}
149
150	0	public void flush() throws IOException {
151		}
152
153	12	public void setAsynchChannelListener(AsynchChannelListener channelListener) {
154	12	this.channelListener = channelListener;
155		}
156
157	0	public AsynchChannelListener getAsynchChannelListener() {
158	0	return channelListener;
159		}
160
161	18	public void dispose() {
162	18	if( running && channelListener!=null ) {
163	14	channelListener.onPacketError(new SocketException("Socket closed."));
164		}
165	18	selection.close();
166	18	super.dispose();
167		}
168
169	12	public void start() throws IOException {
170	12	if( running )
171	0	return;
172	12	running=true;
173	12	selection.setInterestOps(SelectionKey.OP_READ);
174		}
175
176	0	public void stop(long timeout) throws IOException {
177	0	if( !running )
178	0	return;
179	0	running=false;
180	0	selection.setInterestOps(0);
181		}
182		}