View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
4   //  ------------------------------------------------------------------------
5   //  All rights reserved. This program and the accompanying materials
6   //  are made available under the terms of the Eclipse Public License v1.0
7   //  and Apache License v2.0 which accompanies this distribution.
8   //
9   //      The Eclipse Public License is available at
10  //      http://www.eclipse.org/legal/epl-v10.html
11  //
12  //      The Apache License v2.0 is available at
13  //      http://www.opensource.org/licenses/apache2.0.php
14  //
15  //  You may elect to redistribute this code under either of these licenses.
16  //  ========================================================================
17  //
18  
19  package org.eclipse.jetty.websocket.client;
20  
21  import java.io.IOException;
22  import java.net.CookieStore;
23  import java.net.SocketAddress;
24  import java.net.URI;
25  import java.util.ArrayList;
26  import java.util.List;
27  import java.util.Locale;
28  import java.util.concurrent.Executor;
29  import java.util.concurrent.Future;
30  
31  import org.eclipse.jetty.io.ByteBufferPool;
32  import org.eclipse.jetty.io.MappedByteBufferPool;
33  import org.eclipse.jetty.io.SelectorManager;
34  import org.eclipse.jetty.util.HttpCookieStore;
35  import org.eclipse.jetty.util.StringUtil;
36  import org.eclipse.jetty.util.component.ContainerLifeCycle;
37  import org.eclipse.jetty.util.log.Log;
38  import org.eclipse.jetty.util.log.Logger;
39  import org.eclipse.jetty.util.ssl.SslContextFactory;
40  import org.eclipse.jetty.util.thread.QueuedThreadPool;
41  import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
42  import org.eclipse.jetty.util.thread.Scheduler;
43  import org.eclipse.jetty.websocket.api.Session;
44  import org.eclipse.jetty.websocket.api.WebSocketPolicy;
45  import org.eclipse.jetty.websocket.api.extensions.Extension;
46  import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
47  import org.eclipse.jetty.websocket.api.extensions.ExtensionFactory;
48  import org.eclipse.jetty.websocket.client.io.ConnectPromise;
49  import org.eclipse.jetty.websocket.client.io.ConnectionManager;
50  import org.eclipse.jetty.websocket.client.masks.Masker;
51  import org.eclipse.jetty.websocket.client.masks.RandomMasker;
52  import org.eclipse.jetty.websocket.common.events.EventDriver;
53  import org.eclipse.jetty.websocket.common.events.EventDriverFactory;
54  import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
55  
56  /**
57   * WebSocketClient provides a means of establishing connections to remote websocket endpoints.
58   */
59  public class WebSocketClient extends ContainerLifeCycle
60  {
61      private static final Logger LOG = Log.getLogger(WebSocketClient.class);
62  
63      private final WebSocketPolicy policy;
64      private final SslContextFactory sslContextFactory;
65      private final WebSocketExtensionFactory extensionRegistry;
66      private final EventDriverFactory eventDriverFactory;
67      private ByteBufferPool bufferPool;
68      private Executor executor;
69      private Scheduler scheduler;
70      private CookieStore cookieStore;
71      private ConnectionManager connectionManager;
72      private Masker masker;
73      private SocketAddress bindAddress;
74      private long connectTimeout = SelectorManager.DEFAULT_CONNECT_TIMEOUT;
75  
76      public WebSocketClient()
77      {
78          this(null);
79      }
80  
81      public WebSocketClient(SslContextFactory sslContextFactory)
82      {
83          this.sslContextFactory = sslContextFactory;
84          this.policy = WebSocketPolicy.newClientPolicy();
85          this.extensionRegistry = new WebSocketExtensionFactory(policy,bufferPool);
86          this.masker = new RandomMasker();
87          this.eventDriverFactory = new EventDriverFactory(policy);
88      }
89  
90      public Future<Session> connect(Object websocket, URI toUri) throws IOException
91      {
92          ClientUpgradeRequest request = new ClientUpgradeRequest(toUri);
93          request.setRequestURI(toUri);
94          request.setCookiesFrom(this.cookieStore);
95  
96          return connect(websocket,toUri,request);
97      }
98  
99      public Future<Session> connect(Object websocket, URI toUri, ClientUpgradeRequest request) throws IOException
100     {
101         if (!isStarted())
102         {
103             throw new IllegalStateException(WebSocketClient.class.getSimpleName() + "@" + this.hashCode() + " is not started");
104         }
105 
106         // Validate websocket URI
107         if (!toUri.isAbsolute())
108         {
109             throw new IllegalArgumentException("WebSocket URI must be absolute");
110         }
111 
112         if (StringUtil.isBlank(toUri.getScheme()))
113         {
114             throw new IllegalArgumentException("WebSocket URI must include a scheme");
115         }
116 
117         String scheme = toUri.getScheme().toLowerCase(Locale.ENGLISH);
118         if (("ws".equals(scheme) == false) && ("wss".equals(scheme) == false))
119         {
120             throw new IllegalArgumentException("WebSocket URI scheme only supports [ws] and [wss], not [" + scheme + "]");
121         }
122 
123         request.setRequestURI(toUri);
124         request.setCookiesFrom(this.cookieStore);
125 
126         // Validate Requested Extensions
127         for (ExtensionConfig reqExt : request.getExtensions())
128         {
129             if (!extensionRegistry.isAvailable(reqExt.getName()))
130             {
131                 throw new IllegalArgumentException("Requested extension [" + reqExt.getName() + "] is not installed");
132             }
133         }
134 
135         // Validate websocket URI
136         LOG.debug("connect websocket:{} to:{}",websocket,toUri);
137 
138         // Grab Connection Manager
139         ConnectionManager manager = getConnectionManager();
140 
141         // Setup Driver for user provided websocket
142         EventDriver driver = eventDriverFactory.wrap(websocket);
143 
144         // Create the appropriate (physical vs virtual) connection task
145         ConnectPromise promise = manager.connect(this,driver,request);
146 
147         // Execute the connection on the executor thread
148         executor.execute(promise);
149 
150         // Return the future
151         return promise;
152     }
153 
154     @Override
155     protected void doStart() throws Exception
156     {
157         LOG.debug("Starting {}",this);
158 
159         if (sslContextFactory != null)
160         {
161             addBean(sslContextFactory);
162         }
163 
164         String name = WebSocketClient.class.getSimpleName() + "@" + hashCode();
165 
166         if (executor == null)
167         {
168             QueuedThreadPool threadPool = new QueuedThreadPool();
169             threadPool.setName(name);
170             executor = threadPool;
171         }
172         addBean(executor);
173 
174         if (bufferPool == null)
175         {
176             bufferPool = new MappedByteBufferPool();
177         }
178         addBean(bufferPool);
179 
180         if (scheduler == null)
181         {
182             scheduler = new ScheduledExecutorScheduler(name + "-scheduler",false);
183         }
184         addBean(scheduler);
185 
186         if (cookieStore == null)
187         {
188             cookieStore = new HttpCookieStore.Empty();
189         }
190 
191         this.connectionManager = newConnectionManager();
192         addBean(this.connectionManager);
193 
194         super.doStart();
195 
196         LOG.info("Started {}",this);
197     }
198 
199     @Override
200     protected void doStop() throws Exception
201     {
202         LOG.debug("Stopping {}",this);
203 
204         if (cookieStore != null)
205         {
206             cookieStore.removeAll();
207             cookieStore = null;
208         }
209 
210         super.doStop();
211         LOG.info("Stopped {}",this);
212     }
213 
214     public SocketAddress getBindAddress()
215     {
216         return bindAddress;
217     }
218 
219     public ByteBufferPool getBufferPool()
220     {
221         return bufferPool;
222     }
223 
224     public ConnectionManager getConnectionManager()
225     {
226         return connectionManager;
227     }
228 
229     public long getConnectTimeout()
230     {
231         return connectTimeout;
232     }
233 
234     public CookieStore getCookieStore()
235     {
236         return cookieStore;
237     }
238 
239     public Executor getExecutor()
240     {
241         return executor;
242     }
243 
244     public ExtensionFactory getExtensionFactory()
245     {
246         return extensionRegistry;
247     }
248 
249     public Masker getMasker()
250     {
251         return masker;
252     }
253 
254     /**
255      * Get the max idle timeout for new connections.
256      * 
257      * @return the max idle timeout in milliseconds for new connections.
258      */
259     public long getMaxIdleTimeout()
260     {
261         return this.policy.getIdleTimeout();
262     }
263 
264     public WebSocketPolicy getPolicy()
265     {
266         return this.policy;
267     }
268 
269     public Scheduler getScheduler()
270     {
271         return scheduler;
272     }
273 
274     /**
275      * @return the {@link SslContextFactory} that manages TLS encryption
276      * @see WebSocketClient(SslContextFactory)
277      */
278     public SslContextFactory getSslContextFactory()
279     {
280         return sslContextFactory;
281     }
282 
283     public List<Extension> initExtensions(List<ExtensionConfig> requested)
284     {
285         List<Extension> extensions = new ArrayList<Extension>();
286 
287         for (ExtensionConfig cfg : requested)
288         {
289             Extension extension = extensionRegistry.newInstance(cfg);
290 
291             if (extension == null)
292             {
293                 continue;
294             }
295 
296             LOG.debug("added {}",extension);
297             extensions.add(extension);
298         }
299         LOG.debug("extensions={}",extensions);
300         return extensions;
301     }
302 
303     /**
304      * Factory method for new ConnectionManager (used by other projects like cometd)
305      * 
306      * @return the ConnectionManager instance to use
307      */
308     protected ConnectionManager newConnectionManager()
309     {
310         return new ConnectionManager(this);
311     }
312 
313     public void setBindAdddress(SocketAddress bindAddress)
314     {
315         this.bindAddress = bindAddress;
316     }
317 
318     public void setBufferPool(ByteBufferPool bufferPool)
319     {
320         this.bufferPool = bufferPool;
321     }
322 
323     /**
324      * Set the timeout for connecting to the remote server.
325      * 
326      * @param timeoutMilliseconds
327      *            the timeout in milliseconds
328      */
329     public void setConnectTimeout(long timeoutMilliseconds)
330     {
331         if (timeoutMilliseconds < 0)
332         {
333             throw new IllegalStateException("Connect Timeout cannot be negative");
334         }
335         this.connectTimeout = timeoutMilliseconds;
336     }
337 
338     public void setCookieStore(CookieStore cookieStore)
339     {
340         this.cookieStore = cookieStore;
341     }
342 
343     public void setExecutor(Executor executor)
344     {
345         this.executor = executor;
346     }
347 
348     public void setMasker(Masker masker)
349     {
350         this.masker = masker;
351     }
352 
353     /**
354      * Set the max idle timeout for new connections.
355      * <p>
356      * Existing connections will not have their max idle timeout adjusted.
357      * 
358      * @param milliseconds
359      *            the timeout in milliseconds
360      */
361     public void setMaxIdleTimeout(long milliseconds)
362     {
363         this.policy.setIdleTimeout(milliseconds);
364     }
365 }