View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2016 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.common;
20  
21  import java.io.IOException;
22  import java.net.InetSocketAddress;
23  import java.net.URI;
24  import java.util.HashMap;
25  import java.util.Iterator;
26  import java.util.List;
27  import java.util.Map;
28  import java.util.Objects;
29  import java.util.ServiceLoader;
30  import java.util.concurrent.Executor;
31  
32  import org.eclipse.jetty.io.ByteBufferPool;
33  import org.eclipse.jetty.io.Connection;
34  import org.eclipse.jetty.util.annotation.ManagedAttribute;
35  import org.eclipse.jetty.util.annotation.ManagedObject;
36  import org.eclipse.jetty.util.component.ContainerLifeCycle;
37  import org.eclipse.jetty.util.component.Dumpable;
38  import org.eclipse.jetty.util.log.Log;
39  import org.eclipse.jetty.util.log.Logger;
40  import org.eclipse.jetty.util.thread.ThreadClassLoaderScope;
41  import org.eclipse.jetty.websocket.api.BatchMode;
42  import org.eclipse.jetty.websocket.api.CloseException;
43  import org.eclipse.jetty.websocket.api.CloseStatus;
44  import org.eclipse.jetty.websocket.api.RemoteEndpoint;
45  import org.eclipse.jetty.websocket.api.Session;
46  import org.eclipse.jetty.websocket.api.StatusCode;
47  import org.eclipse.jetty.websocket.api.SuspendToken;
48  import org.eclipse.jetty.websocket.api.UpgradeRequest;
49  import org.eclipse.jetty.websocket.api.UpgradeResponse;
50  import org.eclipse.jetty.websocket.api.WebSocketBehavior;
51  import org.eclipse.jetty.websocket.api.WebSocketException;
52  import org.eclipse.jetty.websocket.api.WebSocketPolicy;
53  import org.eclipse.jetty.websocket.api.extensions.ExtensionFactory;
54  import org.eclipse.jetty.websocket.api.extensions.Frame;
55  import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
56  import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
57  import org.eclipse.jetty.websocket.common.events.EventDriver;
58  import org.eclipse.jetty.websocket.common.io.IOState;
59  import org.eclipse.jetty.websocket.common.io.IOState.ConnectionStateListener;
60  import org.eclipse.jetty.websocket.common.scopes.WebSocketContainerScope;
61  import org.eclipse.jetty.websocket.common.scopes.WebSocketSessionScope;
62  
63  @ManagedObject("A Jetty WebSocket Session")
64  public class WebSocketSession extends ContainerLifeCycle implements Session, RemoteEndpointFactory, WebSocketSessionScope, IncomingFrames, Connection.Listener, ConnectionStateListener
65  {
66      private static final Logger LOG = Log.getLogger(WebSocketSession.class);
67      private static final Logger LOG_OPEN = Log.getLogger(WebSocketSession.class.getName() + "_OPEN");
68      private final WebSocketContainerScope containerScope;
69      private final URI requestURI;
70      private final LogicalConnection connection;
71      private final EventDriver websocket;
72      private final Executor executor;
73      private ClassLoader classLoader;
74      private ExtensionFactory extensionFactory;
75      private RemoteEndpointFactory remoteEndpointFactory;
76      private String protocolVersion;
77      private Map<String, String[]> parameterMap = new HashMap<>();
78      private RemoteEndpoint remote;
79      private IncomingFrames incomingHandler;
80      private OutgoingFrames outgoingHandler;
81      private WebSocketPolicy policy;
82      private UpgradeRequest upgradeRequest;
83      private UpgradeResponse upgradeResponse;
84  
85      public WebSocketSession(WebSocketContainerScope containerScope, URI requestURI, EventDriver websocket, LogicalConnection connection)
86      {
87          Objects.requireNonNull(containerScope,"Container Scope cannot be null");
88          Objects.requireNonNull(requestURI,"Request URI cannot be null");
89  
90          this.classLoader = Thread.currentThread().getContextClassLoader();
91          this.containerScope = containerScope;
92          this.requestURI = requestURI;
93          this.websocket = websocket;
94          this.connection = connection;
95          this.executor = connection.getExecutor();
96          this.outgoingHandler = connection;
97          this.incomingHandler = websocket;
98          this.connection.getIOState().addListener(this);
99          this.policy = containerScope.getPolicy();
100 
101         addBean(this.connection);
102         addBean(this.websocket);
103     }
104 
105     @Override
106     public void close()
107     {
108         /* This is assumed to always be a NORMAL closure, no reason phrase */
109         close(StatusCode.NORMAL, null);
110     }
111 
112     @Override
113     public void close(CloseStatus closeStatus)
114     {
115         close(closeStatus.getCode(),closeStatus.getPhrase());
116     }
117 
118     @Override
119     public void close(int statusCode, String reason)
120     {
121         connection.close(statusCode,reason);
122     }
123 
124     /**
125      * Harsh disconnect
126      */
127     @Override
128     public void disconnect()
129     {
130         connection.disconnect();
131 
132         // notify of harsh disconnect
133         notifyClose(StatusCode.NO_CLOSE,"Harsh disconnect");
134     }
135 
136     public void dispatch(Runnable runnable)
137     {
138         executor.execute(runnable);
139     }
140 
141     @Override
142     protected void doStart() throws Exception
143     {
144         if(LOG.isDebugEnabled())
145             LOG.debug("starting - {}",this);
146 
147         Iterator<RemoteEndpointFactory> iter = ServiceLoader.load(RemoteEndpointFactory.class).iterator();
148         if (iter.hasNext())
149             remoteEndpointFactory = iter.next();
150 
151         if (remoteEndpointFactory == null)
152             remoteEndpointFactory = this;
153 
154         if (LOG.isDebugEnabled())
155             LOG.debug("Using RemoteEndpointFactory: {}", remoteEndpointFactory);
156 
157         super.doStart();
158     }
159 
160     @Override
161     protected void doStop() throws Exception
162     {
163         if(LOG.isDebugEnabled())
164             LOG.debug("stopping - {}",this);
165         try
166         {
167             close(StatusCode.SHUTDOWN,"Shutdown");
168         }
169         catch (Throwable t)
170         {
171             LOG.debug("During Connection Shutdown",t);
172         }
173         super.doStop();
174     }
175 
176     @Override
177     public void dump(Appendable out, String indent) throws IOException
178     {
179         dumpThis(out);
180         out.append(indent).append(" +- incomingHandler : ");
181         if (incomingHandler instanceof Dumpable)
182         {
183             ((Dumpable)incomingHandler).dump(out,indent + "    ");
184         }
185         else
186         {
187             out.append(incomingHandler.toString()).append(System.lineSeparator());
188         }
189 
190         out.append(indent).append(" +- outgoingHandler : ");
191         if (outgoingHandler instanceof Dumpable)
192         {
193             ((Dumpable)outgoingHandler).dump(out,indent + "    ");
194         }
195         else
196         {
197             out.append(outgoingHandler.toString()).append(System.lineSeparator());
198         }
199     }
200 
201     @Override
202     public boolean equals(Object obj)
203     {
204         if (this == obj)
205         {
206             return true;
207         }
208         if (obj == null)
209         {
210             return false;
211         }
212         if (getClass() != obj.getClass())
213         {
214             return false;
215         }
216         WebSocketSession other = (WebSocketSession)obj;
217         if (connection == null)
218         {
219             if (other.connection != null)
220             {
221                 return false;
222             }
223         }
224         else if (!connection.equals(other.connection))
225         {
226             return false;
227         }
228         return true;
229     }
230 
231     public ByteBufferPool getBufferPool()
232     {
233         return this.connection.getBufferPool();
234     }
235 
236     public ClassLoader getClassLoader()
237     {
238         return this.getClass().getClassLoader();
239     }
240 
241     public LogicalConnection getConnection()
242     {
243         return connection;
244     }
245 
246     @Override
247     public WebSocketContainerScope getContainerScope()
248     {
249         return this.containerScope;
250     }
251 
252     public ExtensionFactory getExtensionFactory()
253     {
254         return extensionFactory;
255     }
256 
257     /**
258      * The idle timeout in milliseconds
259      */
260     @Override
261     public long getIdleTimeout()
262     {
263         return connection.getMaxIdleTimeout();
264     }
265 
266     @ManagedAttribute(readonly = true)
267     public IncomingFrames getIncomingHandler()
268     {
269         return incomingHandler;
270     }
271 
272     @Override
273     public InetSocketAddress getLocalAddress()
274     {
275         return connection.getLocalAddress();
276     }
277 
278     @ManagedAttribute(readonly = true)
279     public OutgoingFrames getOutgoingHandler()
280     {
281         return outgoingHandler;
282     }
283 
284     @Override
285     public WebSocketPolicy getPolicy()
286     {
287         return policy;
288     }
289 
290     @Override
291     public String getProtocolVersion()
292     {
293         return protocolVersion;
294     }
295 
296     @Override
297     public RemoteEndpoint getRemote()
298     {
299         if(LOG_OPEN.isDebugEnabled())
300             LOG_OPEN.debug("[{}] {}.getRemote()",policy.getBehavior(),this.getClass().getSimpleName());
301         ConnectionState state = connection.getIOState().getConnectionState();
302 
303         if ((state == ConnectionState.OPEN) || (state == ConnectionState.CONNECTED))
304         {
305             return remote;
306         }
307 
308         throw new WebSocketException("RemoteEndpoint unavailable, current state [" + state + "], expecting [OPEN or CONNECTED]");
309     }
310 
311     @Override
312     public InetSocketAddress getRemoteAddress()
313     {
314         return remote.getInetSocketAddress();
315     }
316 
317     public URI getRequestURI()
318     {
319         return requestURI;
320     }
321 
322     @Override
323     public UpgradeRequest getUpgradeRequest()
324     {
325         return this.upgradeRequest;
326     }
327 
328     @Override
329     public UpgradeResponse getUpgradeResponse()
330     {
331         return this.upgradeResponse;
332     }
333 
334 
335     @Override
336     public WebSocketSession getWebSocketSession()
337     {
338         return this;
339     }
340 
341     @Override
342     public int hashCode()
343     {
344         final int prime = 31;
345         int result = 1;
346         result = (prime * result) + ((connection == null)?0:connection.hashCode());
347         return result;
348     }
349 
350     /**
351      * Incoming Errors from Parser
352      */
353     @Override
354     public void incomingError(Throwable t)
355     {
356         if (connection.getIOState().isInputAvailable())
357         {
358             // Forward Errors to User WebSocket Object
359             websocket.incomingError(t);
360         }
361     }
362 
363     /**
364      * Incoming Raw Frames from Parser
365      */
366     @Override
367     public void incomingFrame(Frame frame)
368     {
369         ClassLoader old = Thread.currentThread().getContextClassLoader();
370         try
371         {
372             Thread.currentThread().setContextClassLoader(classLoader);
373             if (connection.getIOState().isInputAvailable())
374             {
375                 // Forward Frames Through Extension List
376                 incomingHandler.incomingFrame(frame);
377             }
378         }
379         finally
380         {
381             Thread.currentThread().setContextClassLoader(old);
382         }
383     }
384 
385     @Override
386     public boolean isOpen()
387     {
388         if (this.connection == null)
389         {
390             return false;
391         }
392         return this.connection.isOpen();
393     }
394 
395     @Override
396     public boolean isSecure()
397     {
398         if (upgradeRequest == null)
399         {
400             throw new IllegalStateException("No valid UpgradeRequest yet");
401         }
402 
403         URI requestURI = upgradeRequest.getRequestURI();
404 
405         return "wss".equalsIgnoreCase(requestURI.getScheme());
406     }
407 
408     public void notifyClose(int statusCode, String reason)
409     {
410         if (LOG.isDebugEnabled())
411         {
412             LOG.debug("notifyClose({},{})",statusCode,reason);
413         }
414         websocket.onClose(new CloseInfo(statusCode,reason));
415     }
416 
417     public void notifyError(Throwable cause)
418     {
419         incomingError(cause);
420     }
421 
422     @Override
423     public void onClosed(Connection connection)
424     {
425     }
426 
427     @Override
428     public void onOpened(Connection connection)
429     {
430         if(LOG_OPEN.isDebugEnabled())
431             LOG_OPEN.debug("[{}] {}.onOpened()",policy.getBehavior(),this.getClass().getSimpleName());
432         open();
433     }
434 
435     @SuppressWarnings("incomplete-switch")
436     @Override
437     public void onConnectionStateChange(ConnectionState state)
438     {
439         switch (state)
440         {
441             case CLOSED:
442                 IOState ioState = this.connection.getIOState();
443                 CloseInfo close = ioState.getCloseInfo();
444                 // confirmed close of local endpoint
445                 notifyClose(close.getStatusCode(),close.getReason());
446                 try
447                 {
448                     if (LOG.isDebugEnabled())
449                         LOG.debug("{}.onSessionClosed()",containerScope.getClass().getSimpleName());
450                     containerScope.onSessionClosed(this);
451                 }
452                 catch (Throwable t)
453                 {
454                     LOG.ignore(t);
455                 }
456                 break;
457             case CONNECTED:
458                 // notify session listeners
459                 try
460                 {
461                     if (LOG.isDebugEnabled())
462                         LOG.debug("{}.onSessionOpened()",containerScope.getClass().getSimpleName());
463                     containerScope.onSessionOpened(this);
464                 }
465                 catch (Throwable t)
466                 {
467                     LOG.ignore(t);
468                 }
469                 break;
470         }
471     }
472 
473     public WebSocketRemoteEndpoint newRemoteEndpoint(LogicalConnection connection, OutgoingFrames outgoingFrames, BatchMode batchMode)
474     {
475         return new WebSocketRemoteEndpoint(connection,outgoingHandler,getBatchMode());
476     }
477 
478     /**
479      * Open/Activate the session
480      */
481     public void open()
482     {
483         if(LOG_OPEN.isDebugEnabled())
484             LOG_OPEN.debug("[{}] {}.open()",policy.getBehavior(),this.getClass().getSimpleName());
485 
486         if (remote != null)
487         {
488             // already opened
489             return;
490         }
491 
492         try(ThreadClassLoaderScope scope = new ThreadClassLoaderScope(classLoader))
493         {
494             // Upgrade success
495             connection.getIOState().onConnected();
496 
497             // Connect remote
498             remote = remoteEndpointFactory.newRemoteEndpoint(connection,outgoingHandler,getBatchMode());
499             if(LOG_OPEN.isDebugEnabled())
500                 LOG_OPEN.debug("[{}] {}.open() remote={}",policy.getBehavior(),this.getClass().getSimpleName(),remote);
501 
502             // Open WebSocket
503             websocket.openSession(this);
504 
505             // Open connection
506             connection.getIOState().onOpened();
507 
508             if (LOG.isDebugEnabled())
509             {
510                 LOG.debug("open -> {}",dump());
511             }
512         }
513         catch (CloseException ce)
514         {
515             LOG.warn(ce);
516             close(ce.getStatusCode(),ce.getMessage());
517         }
518         catch (Throwable t)
519         {
520             LOG.warn(t);
521             // Exception on end-user WS-Endpoint.
522             // Fast-fail & close connection with reason.
523             int statusCode = StatusCode.SERVER_ERROR;
524             if(policy.getBehavior() == WebSocketBehavior.CLIENT)
525             {
526                 statusCode = StatusCode.POLICY_VIOLATION;
527             }
528             close(statusCode,t.getMessage());
529         }
530     }
531 
532     public void setExtensionFactory(ExtensionFactory extensionFactory)
533     {
534         this.extensionFactory = extensionFactory;
535     }
536 
537     /**
538      * Set the timeout in milliseconds
539      */
540     @Override
541     public void setIdleTimeout(long ms)
542     {
543         connection.setMaxIdleTimeout(ms);
544     }
545 
546     public void setOutgoingHandler(OutgoingFrames outgoing)
547     {
548         this.outgoingHandler = outgoing;
549     }
550 
551     public void setPolicy(WebSocketPolicy policy)
552     {
553         this.policy = policy;
554     }
555 
556     public void setUpgradeRequest(UpgradeRequest request)
557     {
558         this.upgradeRequest = request;
559         this.protocolVersion = request.getProtocolVersion();
560         this.parameterMap.clear();
561         if (request.getParameterMap() != null)
562         {
563             for (Map.Entry<String, List<String>> entry : request.getParameterMap().entrySet())
564             {
565                 List<String> values = entry.getValue();
566                 if (values != null)
567                 {
568                     this.parameterMap.put(entry.getKey(),values.toArray(new String[values.size()]));
569                 }
570                 else
571                 {
572                     this.parameterMap.put(entry.getKey(),new String[0]);
573                 }
574             }
575         }
576     }
577 
578     public void setUpgradeResponse(UpgradeResponse response)
579     {
580         this.upgradeResponse = response;
581     }
582 
583     @Override
584     public SuspendToken suspend()
585     {
586         return connection.suspend();
587     }
588 
589     /**
590      * @return the default (initial) value for the batching mode.
591      */
592     public BatchMode getBatchMode()
593     {
594         return BatchMode.AUTO;
595     }
596 
597     @Override
598     public String toString()
599     {
600         StringBuilder builder = new StringBuilder();
601         builder.append("WebSocketSession[");
602         builder.append("websocket=").append(websocket);
603         builder.append(",behavior=").append(policy.getBehavior());
604         builder.append(",connection=").append(connection);
605         builder.append(",remote=").append(remote);
606         builder.append(",incoming=").append(incomingHandler);
607         builder.append(",outgoing=").append(outgoingHandler);
608         builder.append("]");
609         return builder.toString();
610     }
611 
612     public static interface Listener
613     {
614         void onOpened(WebSocketSession session);
615 
616         void onClosed(WebSocketSession session);
617     }
618 }