Clover coverage report - ActiveIO - 1.0 Coverage timestamp: Fri Apr 22 2005 14:27:22 PDT FRAMES NO FRAMES |
file stats: | LOC: | 162 |  | Methods: | 16
| NCLOC: | 132 |  | Classes: | 3
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
AsynchWriteAsynchChannelAdapter.java | 0% | 0% | 0% | 0% |

1 |  |
     package org.activeio.adapter;
2 |  |
3 |  |
     import java.io.IOException;
4 |  |
     import java.io.InterruptedIOException;
5 |  |
6 |  |
     import org.activeio.AsynchChannel;
7 |  |
     import org.activeio.ChannelFactory;
8 |  |
     import org.activeio.FilterAsynchChannel;
9 |  |
     import org.activeio.Packet;
10 |  |
11 |  |
     import EDU.oswego.cs.dl.util.concurrent.BoundedBuffer;
12 |  |
     import EDU.oswego.cs.dl.util.concurrent.Channel;
13 |  |
     import EDU.oswego.cs.dl.util.concurrent.Executor;
14 |  |
     import EDU.oswego.cs.dl.util.concurrent.Latch;
15 |  |
     import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
16 |  |
     import EDU.oswego.cs.dl.util.concurrent.SynchronizedInt;
17 |  |
18 |  |
     public class AsynchWriteAsynchChannelAdapter extends FilterAsynchChannel {
19 |  |
20 |  |
         static public class ObjectDispatcherX implements Runnable {
21 |  |
22 |  |
             private final Executor executor;
23 |  |
             private final Channel queue;
24 |  |
             private final SynchronizedInt size = new SynchronizedInt(0);
25 |  |
             private final AsynchWriteAsynchChannelAdapter objectListener;
26 |  |
             private long pollDelay=10;
27 |  |
28 | 0 |
             public ObjectDispatcherX(AsynchWriteAsynchChannelAdapter objectListener) {
29 | 0 |
                 this(objectListener, 10);
30 |  |
             }
31 |  |
32 | 0 |
             public ObjectDispatcherX(AsynchWriteAsynchChannelAdapter objectListener, int queueSize) {
33 | 0 |
                 this(objectListener, ChannelFactory.DEFAULT_EXECUTOR, new BoundedBuffer(queueSize));
34 |  |
             }
35 |  |
36 | 0 |
             public ObjectDispatcherX(AsynchWriteAsynchChannelAdapter objectListener, Executor executor, Channel queue) {
37 | 0 |
                 this.objectListener = objectListener;
38 | 0 |
                 this.executor = executor;
39 | 0 |
                 this.queue=queue;
40 |  |
             }
41 |  |
42 | 0 |
             public void add(Object o) throws InterruptedException {
43 | 0 |
                 int t = size.increment();
44 | 0 |
                 queue.put(o);
45 | 0 |
                 if( t==1 ) {
46 | 0 |
                     executor.execute(this);
47 |  |
                 }
48 |  |
             }
49 |  |
50 | 0 |
             synchronized public void run() {
51 | 0 |
                 int t = size.get();
52 | 0 |
                 while( t > 0 ) {
53 | 0 |
                     int count=0;
54 | 0 |
                     try {
55 | 0 |
                         Object o;
56 | 0 |
                         while( (o=queue.poll(pollDelay))!=null ) {
57 | 0 |
                             count++;
58 | 0 |
                             objectListener.onObject(o);
59 |  |
                         }
60 |  |
                     } catch (InterruptedException e) {
61 | 0 |
                         Thread.currentThread().interrupt();
62 | 0 |
                         return;
63 |  |
                     } finally {
64 | 0 |
                         t = size.subtract(count);
65 |  |
                     }
66 |  |
                 }
67 |  |
             }
68 |  |
69 |  |
         }
70 |  |
         static public class ObjectDispatcher {
71 |  |
72 |  |
             private final PooledExecutor executor;
73 |  |
             private final AsynchWriteAsynchChannelAdapter objectListener;
74 |  |
75 | 0 |
             public ObjectDispatcher(AsynchWriteAsynchChannelAdapter objectListener) {
76 | 0 |
                 this(objectListener, 10);
77 |  |
             }
78 |  |
79 | 0 |
             public ObjectDispatcher(AsynchWriteAsynchChannelAdapter objectListener, int queueSize) {
80 | 0 |
                 this.objectListener = objectListener;
81 | 0 |
                 executor = new PooledExecutor(new BoundedBuffer(queueSize), 1);
82 | 0 |
                 executor.waitWhenBlocked();
83 |  |
             }
84 |  |
85 | 0 |
             public void add(final Object o) throws InterruptedException {
86 | 0 |
                 executor.execute(new Runnable(){
87 | 0 |
                     public void run() {
88 | 0 |
                         objectListener.onObject(o);
89 |  |
                     }
90 |  |
                 });
91 |  |
             }
92 |  |
         }
93 |  |
94 |  |
         private final ObjectDispatcher dispatcher;
95 |  |
         private static final Object FLUSH_COMMAND = new Object();
96 |  |
97 | 0 |
         public AsynchWriteAsynchChannelAdapter(AsynchChannel next) {
98 | 0 |
             this(next, 10);
99 |  |
         }
100 |  |
101 | 0 |
         public AsynchWriteAsynchChannelAdapter(AsynchChannel next, int queueSize) {
102 | 0 |
             super(next);
103 | 0 |
             this.dispatcher = new ObjectDispatcher(this, queueSize);
104 |  |
         }
105 |  |
106 | 0 |
         public void onObject(Object o) {
107 | 0 |
             try {
108 | 0 |
                 if( o == FLUSH_COMMAND ) {
109 | 0 |
                     next.flush();
110 | 0 |
                     return;
111 |  |
                 }
112 | 0 |
                 if( o.getClass() == Latch.class ) {
113 | 0 |
                     next.flush();
114 | 0 |
                     ((Latch)o).release();
115 | 0 |
                     return;
116 |  |
                 }
117 | 0 |
                 next.write((Packet)o);
118 |  |
             } catch (IOException e) {
119 | 0 |
                 channelListener.onPacketError(e);
120 |  |
             }
121 |  |
         }
122 |  |
123 | 0 |
         public void write(Packet packet) throws IOException {
124 | 0 |
             try {
125 | 0 |
                 dispatcher.add(packet);
126 |  |
             } catch (InterruptedException e) {
127 | 0 |
                 throw new InterruptedIOException();
128 |  |
             }
129 |  |
         }
130 |  |
131 | 0 |
         public void flush() throws IOException {
132 | 0 |
             flush(NO_WAIT_TIMEOUT);
133 |  |
         }
134 |  |
135 | 0 |
         public void stop(long timeout) throws IOException {
136 | 0 |
             flush(WAIT_FOREVER_TIMEOUT);
137 |  |
         }
138 |  |
139 |  |
140 |  |
         /**
141 |  |
          * @param timeout
142 |  |
          * @throws InterruptedIOException
143 |  |
          */
144 | 0 |
         private void flush(long timeout) throws InterruptedIOException {
145 | 0 |
             try {
146 | 0 |
                 if( timeout == NO_WAIT_TIMEOUT ) {
147 | 0 |
                     dispatcher.add(FLUSH_COMMAND);
148 | 0 |
                 } else if( timeout == WAIT_FOREVER_TIMEOUT ) {
149 | 0 |
                     Latch l = new Latch();
150 | 0 |
                     dispatcher.add(l);
151 | 0 |
                     l.acquire();
152 |  |
                 } else {
153 | 0 |
                     Latch l = new Latch();
154 | 0 |
                     dispatcher.add(l);
155 | 0 |
                     l.attempt(timeout);
156 |  |
                 }
157 |  |
             } catch (InterruptedException e) {
158 | 0 |
                 throw new InterruptedIOException();
159 |  |
             }
160 |  |
         }
161 |  |
     }
162 |  |