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:	228		Methods:	15
	NCLOC:	141		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

SynchToAsynchChannelAdapter.java

56.2%

66.2%

80%

65.2%

 /** 
  * 
  * Copyright 2004 Protique Ltd
  * 
  * 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.adapter;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
 
 import org.activeio.AsynchChannel;
 import org.activeio.AsynchChannelListener;
 import org.activeio.Channel;
 import org.activeio.ChannelFactory;
 import org.activeio.Packet;
 import org.activeio.Service;
 import org.activeio.SynchChannel;
 import org.activeio.packet.EOSPacket;
 
 import EDU.oswego.cs.dl.util.concurrent.Executor;
 import EDU.oswego.cs.dl.util.concurrent.Latch;
 import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
 
 /**
  * Adapts a {@see org.activeio.SynchChannel} so that it provides an 
  * {@see org.activeio.AsynchChannel} interface.  When this channel
  * is started, a background thread is used to poll the {@see org.activeio.SynchChannel}
  *  for packets comming up the channel which are then delivered to the 
  * {@see org.activeio.ChannelConsumer}.
  * 
  * @version $Revision$
  */
 final public class SynchToAsynchChannelAdapter implements AsynchChannel, Runnable {
 
     private final SynchronizedBoolean running = new SynchronizedBoolean(false);
     private final SynchChannel synchChannel;
     private final Executor executor;
     private AsynchChannelListener channelListener;
     private Latch doneLatch;
     
     
     static public AsynchChannel adapt(Channel channel) {
         return adapt(channel, ChannelFactory.DEFAULT_EXECUTOR);
     }
 
     static public AsynchChannel adapt(Channel channel, Executor executor) {
         
         // It might not need adapting
         if( channel instanceof AsynchChannel ) {
             return (AsynchChannel) channel;
         }
         
         // Can we just just undo the adaptor
         if( channel.getClass() == SynchToAsynchChannelAdapter.class ) {
             return ((AsynchToSynchChannelAdapter)channel).getAsynchChannel();
         }
         
         return new SynchToAsynchChannelAdapter((SynchChannel) channel, executor);
         
     }
 
     
     /**
      * @deprecated {@see #adapt(SynchChannel)}
      */
     public SynchToAsynchChannelAdapter(SynchChannel synchChannel) {
         this(synchChannel, ChannelFactory.DEFAULT_EXECUTOR);
     }
 
     /**
      * @deprecated {@see #adapt(SynchChannel, Executor)}
      */
     public SynchToAsynchChannelAdapter(SynchChannel synchChannel, Executor executor) {
         this.synchChannel = synchChannel;
         this.executor = executor;
     }
 
     synchronized public void start() throws IOException {
         if (running.commit(false, true)) {
             
             if (channelListener == null)
                 throw new IllegalStateException("UpPacketListener must be set before object can be started.");
             
             synchChannel.start();
 
             try {
                 doneLatch = new Latch();
                 executor.execute(this);
             } catch (InterruptedException e) {
                 throw new InterruptedIOException(e.getMessage());
             }
 
         }
     }
 
     synchronized public void stop(long timeout) throws IOException {
         if (running.commit(true, false)) {
             try {
                 if( timeout == NO_WAIT_TIMEOUT ) {
                     synchChannel.stop(NO_WAIT_TIMEOUT);
                 } else if( timeout == WAIT_FOREVER_TIMEOUT ) {
                     doneLatch.acquire();
                     synchChannel.stop(WAIT_FOREVER_TIMEOUT);
                 } else {
                     
                     long start = System.currentTimeMillis();
                     if( doneLatch.attempt(timeout) ) {
                         timeout -= (System.currentTimeMillis() - start);
                     } else {
                         timeout=0;
                     }
                     
                     if( timeout <= 0 ) {
                         synchChannel.stop(NO_WAIT_TIMEOUT);
                     } else {
                         synchChannel.stop(timeout);
                     }
                 }
             } catch (IOException e) {
                 throw e;
             } catch (Throwable e) {
                 throw (IOException)new IOException("stop failed: " + e.getMessage()).initCause(e);
             }
         }
     }
 
     /**
      * reads packets from a Socket
      */
     public void run() {
         
         // Change the thread name.
         String oldName = Thread.currentThread().getName();        
         Thread.currentThread().setName( synchChannel.toString() );        
         try {
             while (running.get()) {
                 try {
                     Packet packet = synchChannel.read(500);
                     if( packet==null )
                         continue;    
                     
                     if( packet == EOSPacket.EOS_PACKET ) {
                         channelListener.onPacket(packet);
                         return;
                     }
                     
                     if( packet.hasRemaining() ) {
                         channelListener.onPacket(packet);
                     }
                     
                 } catch (IOException e) {
                     channelListener.onPacketError(e);
                 } catch (Throwable e) {
                     channelListener.onPacketError((IOException)new IOException("Unexpected Error: "+e).initCause(e));
                 }
             }
         } finally {
             if( doneLatch!=null )
                 doneLatch.release();
             Thread.currentThread().setName(oldName);
         }
     }
 
     /**
      * @see org.activeio.AsynchChannel#setAsynchChannelListener(org.activeio.UpPacketListener)
      */
     public void setAsynchChannelListener(AsynchChannelListener channelListener) {
         if (running.get())
             throw new IllegalStateException("Cannot change the UpPacketListener while the object is running.");
         this.channelListener = channelListener;
     }
 
     /**
      * @see org.activeio.Channel#write(org.activeio.channel.Packet)
      */
     public void write(org.activeio.Packet packet) throws IOException {
         synchChannel.write(packet);
     }
 
     /**
      * @see org.activeio.Channel#flush()
      */
     public void flush() throws IOException {
         synchChannel.flush();
     }
 
     /**
      * @see org.activeio.Disposable#dispose()
      */
     public void dispose() {
         try {
             stop(Service.NO_WAIT_TIMEOUT);
         } catch ( IOException ignore) {
         }
         synchChannel.dispose();        
     }
 
     public AsynchChannelListener getAsynchChannelListener() {
         return channelListener;
     }
     
     public Object narrow(Class target) {
         if( target.isAssignableFrom(getClass()) ) {
             return this;
         }
         return synchChannel.narrow(target);
     }    
     
     public SynchChannel getSynchChannel() {
         return synchChannel;
     }
 
     public String toString() {
         return synchChannel.toString();
     }
 }

1		/**
2		*
3		* Copyright 2004 Protique Ltd
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		package org.activeio.adapter;
19
20		import java.io.IOException;
21		import java.io.InterruptedIOException;
22
23		import org.activeio.AsynchChannel;
24		import org.activeio.AsynchChannelListener;
25		import org.activeio.Channel;
26		import org.activeio.ChannelFactory;
27		import org.activeio.Packet;
28		import org.activeio.Service;
29		import org.activeio.SynchChannel;
30		import org.activeio.packet.EOSPacket;
31
32		import EDU.oswego.cs.dl.util.concurrent.Executor;
33		import EDU.oswego.cs.dl.util.concurrent.Latch;
34		import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
35
36		/**
37		* Adapts a {@see org.activeio.SynchChannel} so that it provides an
38		* {@see org.activeio.AsynchChannel} interface. When this channel
39		* is started, a background thread is used to poll the {@see org.activeio.SynchChannel}
40		* for packets comming up the channel which are then delivered to the
41		* {@see org.activeio.ChannelConsumer}.
42		*
43		* @version $Revision$
44		*/
45		final public class SynchToAsynchChannelAdapter implements AsynchChannel, Runnable {
46
47		private final SynchronizedBoolean running = new SynchronizedBoolean(false);
48		private final SynchChannel synchChannel;
49		private final Executor executor;
50		private AsynchChannelListener channelListener;
51		private Latch doneLatch;
52
53
54	44	static public AsynchChannel adapt(Channel channel) {
55	44	return adapt(channel, ChannelFactory.DEFAULT_EXECUTOR);
56		}
57
58	46	static public AsynchChannel adapt(Channel channel, Executor executor) {
59
60		// It might not need adapting
61	46	if( channel instanceof AsynchChannel ) {
62	4	return (AsynchChannel) channel;
63		}
64
65		// Can we just just undo the adaptor
66	41	if( channel.getClass() == SynchToAsynchChannelAdapter.class ) {
67	0	return ((AsynchToSynchChannelAdapter)channel).getAsynchChannel();
68		}
69
70	41	return new SynchToAsynchChannelAdapter((SynchChannel) channel, executor);
71
72		}
73
74
75		/**
76		* @deprecated {@see #adapt(SynchChannel)}
77		*/
78	0	public SynchToAsynchChannelAdapter(SynchChannel synchChannel) {
79	0	this(synchChannel, ChannelFactory.DEFAULT_EXECUTOR);
80		}
81
82		/**
83		* @deprecated {@see #adapt(SynchChannel, Executor)}
84		*/
85	42	public SynchToAsynchChannelAdapter(SynchChannel synchChannel, Executor executor) {
86	42	this.synchChannel = synchChannel;
87	42	this.executor = executor;
88		}
89
90	26	synchronized public void start() throws IOException {
91	26	if (running.commit(false, true)) {
92
93	26	if (channelListener == null)
94	0	throw new IllegalStateException("UpPacketListener must be set before object can be started.");
95
96	26	synchChannel.start();
97
98	26	try {
99	26	doneLatch = new Latch();
100	26	executor.execute(this);
101		} catch (InterruptedException e) {
102	0	throw new InterruptedIOException(e.getMessage());
103		}
104
105		}
106		}
107
108	22	synchronized public void stop(long timeout) throws IOException {
109	22	if (running.commit(true, false)) {
110	12	try {
111	12	if( timeout == NO_WAIT_TIMEOUT ) {
112	12	synchChannel.stop(NO_WAIT_TIMEOUT);
113	0	} else if( timeout == WAIT_FOREVER_TIMEOUT ) {
114	0	doneLatch.acquire();
115	0	synchChannel.stop(WAIT_FOREVER_TIMEOUT);
116		} else {
117
118	0	long start = System.currentTimeMillis();
119	0	if( doneLatch.attempt(timeout) ) {
120	0	timeout -= (System.currentTimeMillis() - start);
121		} else {
122	0	timeout=0;
123		}
124
125	0	if( timeout <= 0 ) {
126	0	synchChannel.stop(NO_WAIT_TIMEOUT);
127		} else {
128	0	synchChannel.stop(timeout);
129		}
130		}
131		} catch (IOException e) {
132	0	throw e;
133		} catch (Throwable e) {
134	0	throw (IOException)new IOException("stop failed: " + e.getMessage()).initCause(e);
135		}
136		}
137		}
138
139		/**
140		* reads packets from a Socket
141		*/
142	26	public void run() {
143
144		// Change the thread name.
145	26	String oldName = Thread.currentThread().getName();
146	26	Thread.currentThread().setName( synchChannel.toString() );
147	26	try {
148	26	while (running.get()) {
149	101	try {
150	101	Packet packet = synchChannel.read(500);
151	101	if( packet==null )
152	54	continue;
153
154	47	if( packet == EOSPacket.EOS_PACKET ) {
155	14	channelListener.onPacket(packet);
156	14	return;
157		}
158
159	33	if( packet.hasRemaining() ) {
160	33	channelListener.onPacket(packet);
161		}
162
163		} catch (IOException e) {
164	0	channelListener.onPacketError(e);
165		} catch (Throwable e) {
166	0	channelListener.onPacketError((IOException)new IOException("Unexpected Error: "+e).initCause(e));
167		}
168		}
169		} finally {
170	26	if( doneLatch!=null )
171	26	doneLatch.release();
172	26	Thread.currentThread().setName(oldName);
173		}
174		}
175
176		/**
177		* @see org.activeio.AsynchChannel#setAsynchChannelListener(org.activeio.UpPacketListener)
178		*/
179	38	public void setAsynchChannelListener(AsynchChannelListener channelListener) {
180	38	if (running.get())
181	0	throw new IllegalStateException("Cannot change the UpPacketListener while the object is running.");
182	38	this.channelListener = channelListener;
183		}
184
185		/**
186		* @see org.activeio.Channel#write(org.activeio.channel.Packet)
187		*/
188	24	public void write(org.activeio.Packet packet) throws IOException {
189	24	synchChannel.write(packet);
190		}
191
192		/**
193		* @see org.activeio.Channel#flush()
194		*/
195	10	public void flush() throws IOException {
196	10	synchChannel.flush();
197		}
198
199		/**
200		* @see org.activeio.Disposable#dispose()
201		*/
202	10	public void dispose() {
203	10	try {
204	10	stop(Service.NO_WAIT_TIMEOUT);
205		} catch ( IOException ignore) {
206		}
207	10	synchChannel.dispose();
208		}
209
210	0	public AsynchChannelListener getAsynchChannelListener() {
211	0	return channelListener;
212		}
213
214	8	public Object narrow(Class target) {
215	8	if( target.isAssignableFrom(getClass()) ) {
216	0	return this;
217		}
218	8	return synchChannel.narrow(target);
219		}
220
221	16	public SynchChannel getSynchChannel() {
222	16	return synchChannel;
223		}
224
225	0	public String toString() {
226	0	return synchChannel.toString();
227		}
228		}