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