View Javadoc

1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  
28  package org.apache.http.impl.nio.reactor;
29  
30  import java.io.IOException;
31  import java.net.InetSocketAddress;
32  import java.net.Socket;
33  import java.net.SocketAddress;
34  import java.nio.channels.ByteChannel;
35  import java.nio.channels.SelectionKey;
36  import java.nio.channels.SocketChannel;
37  import java.util.Collections;
38  import java.util.HashMap;
39  import java.util.Map;
40  
41  import org.apache.http.annotation.ThreadSafe;
42  import org.apache.http.nio.reactor.IOSession;
43  import org.apache.http.nio.reactor.SessionBufferStatus;
44  import org.apache.http.nio.reactor.SocketAccessor;
45  import org.apache.http.util.Args;
46  
47  /**
48   * Default implementation of {@link IOSession}.
49   *
50   * @since 4.0
51   */
52  @ThreadSafe
53  public class IOSessionImpl implements IOSession, SocketAccessor {
54  
55      private final SelectionKey key;
56      private final ByteChannel channel;
57      private final Map<String, Object> attributes;
58      private final InterestOpsCallback interestOpsCallback;
59      private final SessionClosedCallback sessionClosedCallback;
60  
61      private volatile int status;
62      private volatile int currentEventMask;
63      private volatile SessionBufferStatus bufferStatus;
64      private volatile int socketTimeout;
65  
66      private final long startedTime;
67  
68      private long lastReadTime;
69      private long lastWriteTime;
70      private long lastAccessTime;
71  
72      /**
73       * Creates new instance of IOSessionImpl.
74       *
75       * @param key the selection key.
76       * @param interestOpsCallback interestOps callback.
77       * @param sessionClosedCallback session closed callback.
78       *
79       * @since 4.1
80       */
81      public IOSessionImpl(
82              final SelectionKey key,
83              final InterestOpsCallback interestOpsCallback,
84              final SessionClosedCallback sessionClosedCallback) {
85          super();
86          Args.notNull(key, "Selection key");
87          this.key = key;
88          this.channel = (ByteChannel) this.key.channel();
89          this.interestOpsCallback = interestOpsCallback;
90          this.sessionClosedCallback = sessionClosedCallback;
91          this.attributes = Collections.synchronizedMap(new HashMap<String, Object>());
92          this.currentEventMask = key.interestOps();
93          this.socketTimeout = 0;
94          this.status = ACTIVE;
95          final long now = System.currentTimeMillis();
96          this.startedTime = now;
97          this.lastReadTime = now;
98          this.lastWriteTime = now;
99          this.lastAccessTime = now;
100     }
101 
102     /**
103      * Creates new instance of IOSessionImpl.
104      *
105      * @param key the selection key.
106      * @param sessionClosedCallback session closed callback.
107      */
108     public IOSessionImpl(
109             final SelectionKey key,
110             final SessionClosedCallback sessionClosedCallback) {
111         this(key, null, sessionClosedCallback);
112     }
113 
114     @Override
115     public ByteChannel channel() {
116         return this.channel;
117     }
118 
119     @Override
120     public SocketAddress getLocalAddress() {
121         if (this.channel instanceof SocketChannel) {
122             return ((SocketChannel)this.channel).socket().getLocalSocketAddress();
123         } else {
124             return null;
125         }
126     }
127 
128     @Override
129     public SocketAddress getRemoteAddress() {
130         if (this.channel instanceof SocketChannel) {
131             return ((SocketChannel)this.channel).socket().getRemoteSocketAddress();
132         } else {
133             return null;
134         }
135     }
136 
137     @Override
138     public synchronized int getEventMask() {
139         return this.interestOpsCallback != null ? this.currentEventMask : this.key.interestOps();
140     }
141 
142     @Override
143     public synchronized void setEventMask(final int ops) {
144         if (this.status == CLOSED) {
145             return;
146         }
147         if (this.interestOpsCallback != null) {
148             // update the current event mask
149             this.currentEventMask = ops;
150 
151             // local variable
152             final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
153 
154             // add this operation to the interestOps() queue
155             this.interestOpsCallback.addInterestOps(entry);
156         } else {
157             this.key.interestOps(ops);
158         }
159         this.key.selector().wakeup();
160     }
161 
162     @Override
163     public synchronized void setEvent(final int op) {
164         if (this.status == CLOSED) {
165             return;
166         }
167         if (this.interestOpsCallback != null) {
168             // update the current event mask
169             this.currentEventMask |= op;
170 
171             // local variable
172             final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
173 
174             // add this operation to the interestOps() queue
175             this.interestOpsCallback.addInterestOps(entry);
176         } else {
177             final int ops = this.key.interestOps();
178             this.key.interestOps(ops | op);
179         }
180         this.key.selector().wakeup();
181     }
182 
183     @Override
184     public synchronized void clearEvent(final int op) {
185         if (this.status == CLOSED) {
186             return;
187         }
188         if (this.interestOpsCallback != null) {
189             // update the current event mask
190             this.currentEventMask &= ~op;
191 
192             // local variable
193             final InterestOpEntry entry = new InterestOpEntry(this.key, this.currentEventMask);
194 
195             // add this operation to the interestOps() queue
196             this.interestOpsCallback.addInterestOps(entry);
197         } else {
198             final int ops = this.key.interestOps();
199             this.key.interestOps(ops & ~op);
200         }
201         this.key.selector().wakeup();
202     }
203 
204     @Override
205     public int getSocketTimeout() {
206         return this.socketTimeout;
207     }
208 
209     @Override
210     public synchronized void setSocketTimeout(final int timeout) {
211         this.socketTimeout = timeout;
212         this.lastAccessTime = System.currentTimeMillis();
213     }
214 
215     @Override
216     public synchronized void close() {
217         if (this.status == CLOSED) {
218             return;
219         }
220         this.status = CLOSED;
221         this.key.cancel();
222         try {
223             this.key.channel().close();
224         } catch (final IOException ex) {
225             // Munching exceptions is not nice
226             // but in this case it is justified
227         }
228         if (this.sessionClosedCallback != null) {
229             this.sessionClosedCallback.sessionClosed(this);
230         }
231         if (this.key.selector().isOpen()) {
232             this.key.selector().wakeup();
233         }
234     }
235 
236     @Override
237     public int getStatus() {
238         return this.status;
239     }
240 
241     @Override
242     public boolean isClosed() {
243         return this.status == CLOSED;
244     }
245 
246     @Override
247     public void shutdown() {
248         // For this type of session, a close() does exactly
249         // what we need and nothing more.
250         close();
251     }
252 
253     @Override
254     public boolean hasBufferedInput() {
255         final SessionBufferStatus buffStatus = this.bufferStatus;
256         return buffStatus != null && buffStatus.hasBufferedInput();
257     }
258 
259     @Override
260     public boolean hasBufferedOutput() {
261         final SessionBufferStatus buffStatus = this.bufferStatus;
262         return buffStatus != null && buffStatus.hasBufferedOutput();
263     }
264 
265     @Override
266     public void setBufferStatus(final SessionBufferStatus bufferStatus) {
267         this.bufferStatus = bufferStatus;
268     }
269 
270     @Override
271     public Object getAttribute(final String name) {
272         return this.attributes.get(name);
273     }
274 
275     @Override
276     public Object removeAttribute(final String name) {
277         return this.attributes.remove(name);
278     }
279 
280     @Override
281     public void setAttribute(final String name, final Object obj) {
282         this.attributes.put(name, obj);
283     }
284 
285     public synchronized long getStartedTime() {
286         return this.startedTime;
287     }
288 
289     public synchronized long getLastReadTime() {
290         return this.lastReadTime;
291     }
292 
293     public synchronized long getLastWriteTime() {
294         return this.lastWriteTime;
295     }
296 
297     public synchronized long getLastAccessTime() {
298         return this.lastAccessTime;
299     }
300 
301     synchronized void resetLastRead() {
302         final long now = System.currentTimeMillis();
303         this.lastReadTime = now;
304         this.lastAccessTime = now;
305     }
306 
307     synchronized void resetLastWrite() {
308         final long now = System.currentTimeMillis();
309         this.lastWriteTime = now;
310         this.lastAccessTime = now;
311     }
312 
313     private static void formatOps(final StringBuilder buffer, final int ops) {
314         if ((ops & SelectionKey.OP_READ) > 0) {
315             buffer.append('r');
316         }
317         if ((ops & SelectionKey.OP_WRITE) > 0) {
318             buffer.append('w');
319         }
320         if ((ops & SelectionKey.OP_ACCEPT) > 0) {
321             buffer.append('a');
322         }
323         if ((ops & SelectionKey.OP_CONNECT) > 0) {
324             buffer.append('c');
325         }
326     }
327 
328     private static void formatAddress(final StringBuilder buffer, final SocketAddress socketAddress) {
329         if (socketAddress instanceof InetSocketAddress) {
330             final InetSocketAddress addr = ((InetSocketAddress) socketAddress);
331             buffer.append(addr.getAddress() != null ? addr.getAddress().getHostAddress() :
332                 addr.getAddress())
333             .append(':')
334             .append(addr.getPort());
335         } else {
336             buffer.append(socketAddress);
337         }
338     }
339 
340     @Override
341     public synchronized String toString() {
342         final StringBuilder buffer = new StringBuilder();
343         final SocketAddress remoteAddress = getRemoteAddress();
344         final SocketAddress localAddress = getLocalAddress();
345         if (remoteAddress != null && localAddress != null) {
346             formatAddress(buffer, localAddress);
347             buffer.append("<->");
348             formatAddress(buffer, remoteAddress);
349         }
350         buffer.append("[");
351         switch (this.status) {
352         case ACTIVE:
353             buffer.append("ACTIVE");
354             break;
355         case CLOSING:
356             buffer.append("CLOSING");
357             break;
358         case CLOSED:
359             buffer.append("CLOSED");
360             break;
361         }
362         buffer.append("][");
363         if (this.key.isValid()) {
364             formatOps(buffer, this.interestOpsCallback != null ?
365                     this.currentEventMask : this.key.interestOps());
366             buffer.append(":");
367             formatOps(buffer, this.key.readyOps());
368         }
369         buffer.append("]");
370         return buffer.toString();
371     }
372 
373     @Override
374     public Socket getSocket() {
375         if (this.channel instanceof SocketChannel) {
376             return ((SocketChannel) this.channel).socket();
377         } else {
378             return null;
379         }
380     }
381 
382 }