View Javadoc
1   /*
2    *  Licensed to the Apache Software Foundation (ASF) under one
3    *  or more contributor license agreements.  See the NOTICE file
4    *  distributed with this work for additional information
5    *  regarding copyright ownership.  The ASF licenses this file
6    *  to you under the Apache License, Version 2.0 (the
7    *  "License"); you may not use this file except in compliance
8    *  with the License.  You may obtain a copy of the License at
9    *
10   *    http://www.apache.org/licenses/LICENSE-2.0
11   *
12   *  Unless required by applicable law or agreed to in writing,
13   *  software distributed under the License is distributed on an
14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   *  KIND, either express or implied.  See the License for the
16   *  specific language governing permissions and limitations
17   *  under the License.
18   *
19   */
20  package org.apache.mina.proxy;
21  
22  import java.util.LinkedList;
23  import java.util.Queue;
24  
25  import org.apache.mina.core.buffer.IoBuffer;
26  import org.apache.mina.core.filterchain.IoFilter.NextFilter;
27  import org.apache.mina.core.future.DefaultWriteFuture;
28  import org.apache.mina.core.future.WriteFuture;
29  import org.apache.mina.core.session.IoSession;
30  import org.apache.mina.core.write.DefaultWriteRequest;
31  import org.apache.mina.core.write.WriteRequest;
32  import org.apache.mina.proxy.filter.ProxyFilter;
33  import org.apache.mina.proxy.filter.ProxyHandshakeIoBuffer;
34  import org.apache.mina.proxy.session.ProxyIoSession;
35  import org.slf4j.Logger;
36  import org.slf4j.LoggerFactory;
37  
38  /**
39   * AbstractProxyLogicHandler.java - Helper class to handle proxy handshaking logic. Derived classes 
40   * implement proxy type specific logic.
41   * <p>
42   * Based upon SSLHandler from mina-filter-ssl.
43   * 
44   * @author <a href="http://mina.apache.org">Apache MINA Project</a>
45   * @since MINA 2.0.0-M3
46   */
47  public abstract class AbstractProxyLogicHandler implements ProxyLogicHandler {
48  
49      private final static Logger LOGGER = LoggerFactory.getLogger(AbstractProxyLogicHandler.class);
50  
51      /**
52       * Object that contains all the proxy authentication session informations.
53       */
54      private ProxyIoSession proxyIoSession;
55  
56      /**
57       * Queue of write events which occurred before the proxy handshake had completed.
58       */
59      private Queue<Event> writeRequestQueue = null;
60  
61      /**
62       * Has the handshake been completed.
63       */
64      private boolean handshakeComplete = false;
65  
66      /**
67       * Creates a new {@link AbstractProxyLogicHandler}.
68       * 
69       * @param proxyIoSession {@link ProxyIoSession} in use.
70       */
71      public AbstractProxyLogicHandler(ProxyIoSession proxyIoSession) {
72          this.proxyIoSession = proxyIoSession;
73      }
74  
75      /**
76       * @return the proxy filter {@link ProxyFilter}.
77       */
78      protected ProxyFilter getProxyFilter() {
79          return proxyIoSession.getProxyFilter();
80      }
81  
82      /**
83       * @return the session.
84       */
85      protected IoSession getSession() {
86          return proxyIoSession.getSession();
87      }
88  
89      /**
90       * @return the {@link ProxyIoSession} object.
91       */
92      public ProxyIoSession getProxyIoSession() {
93          return proxyIoSession;
94      }
95  
96      /**
97       * Writes data to the proxy server.
98       * 
99       * @param nextFilter the next filter
100      * @param data Data buffer to be written.
101      * @return A Future for the write operation
102      */
103     protected WriteFuture writeData(final NextFilter nextFilter, final IoBuffer data) {
104         // write net data
105         ProxyHandshakeIoBuffer writeBuffer = new ProxyHandshakeIoBuffer(data);
106 
107         LOGGER.debug("   session write: {}", writeBuffer);
108 
109         WriteFuture writeFuture = new DefaultWriteFuture(getSession());
110         getProxyFilter().writeData(nextFilter, getSession(), new DefaultWriteRequest(writeBuffer, writeFuture), true);
111 
112         return writeFuture;
113     }
114 
115     /**
116      * @return <tt>true</tt> if handshaking is complete and
117      * data can be sent through the proxy.
118      */
119     public boolean isHandshakeComplete() {
120         synchronized (this) {
121             return handshakeComplete;
122         }
123     }
124 
125     /**
126      * Signals that the handshake has finished.
127      */
128     protected final void setHandshakeComplete() {
129         synchronized (this) {
130             handshakeComplete = true;
131         }
132 
133         ProxyIoSession proxyIoSession = getProxyIoSession();
134         proxyIoSession.getConnector().fireConnected(proxyIoSession.getSession()).awaitUninterruptibly();
135 
136         LOGGER.debug("  handshake completed");
137 
138         // Connected OK
139         try {
140             proxyIoSession.getEventQueue().flushPendingSessionEvents();
141             flushPendingWriteRequests();
142         } catch (Exception ex) {
143             LOGGER.error("Unable to flush pending write requests", ex);
144         }
145     }
146 
147     /**
148      * Send any write requests which were queued whilst waiting for handshaking to complete.
149      * 
150      * @throws Exception If we can't flush the pending write requests
151      */
152     protected synchronized void flushPendingWriteRequests() throws Exception {
153         LOGGER.debug(" flushPendingWriteRequests()");
154 
155         if (writeRequestQueue == null) {
156             return;
157         }
158 
159         Event scheduledWrite;
160         while ((scheduledWrite = writeRequestQueue.poll()) != null) {
161             LOGGER.debug(" Flushing buffered write request: {}", scheduledWrite.data);
162 
163             getProxyFilter().filterWrite(scheduledWrite.nextFilter, getSession(), (WriteRequest) scheduledWrite.data);
164         }
165 
166         // Free queue
167         writeRequestQueue = null;
168     }
169 
170     /**
171      * Enqueue a message to be written once handshaking is complete.
172      */
173     public synchronized void enqueueWriteRequest(final NextFilter nextFilter, final WriteRequest writeRequest) {
174         if (writeRequestQueue == null) {
175             writeRequestQueue = new LinkedList<Event>();
176         }
177 
178         writeRequestQueue.offer(new Event(nextFilter, writeRequest));
179     }
180 
181     /**
182      * Closes the session.
183      * 
184      * @param message the error message
185      * @param t the exception which caused the session closing
186      */
187     protected void closeSession(final String message, final Throwable t) {
188         if (t != null) {
189             LOGGER.error(message, t);
190             proxyIoSession.setAuthenticationFailed(true);
191         } else {
192             LOGGER.error(message);
193         }
194 
195         getSession().close(true);
196     }
197 
198     /**
199      * Closes the session.
200      * 
201      * @param message the error message
202      */
203     protected void closeSession(final String message) {
204         closeSession(message, null);
205     }
206 
207     /**
208      * Event wrapper class for enqueued events.
209      */
210     private final static class Event {
211         private final NextFilter nextFilter;
212 
213         private final Object data;
214 
215         Event(final NextFilter nextFilter, final Object data) {
216             this.nextFilter = nextFilter;
217             this.data = data;
218         }
219     }
220 }