View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2015 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.jsr356.endpoints;
20  
21  import java.io.IOException;
22  import java.io.InputStream;
23  import java.io.Reader;
24  import java.nio.ByteBuffer;
25  import java.util.Map;
26  
27  import javax.websocket.CloseReason;
28  import javax.websocket.Endpoint;
29  import javax.websocket.MessageHandler;
30  import javax.websocket.MessageHandler.Whole;
31  import javax.websocket.PongMessage;
32  
33  import org.eclipse.jetty.util.BufferUtil;
34  import org.eclipse.jetty.util.log.Log;
35  import org.eclipse.jetty.util.log.Logger;
36  import org.eclipse.jetty.websocket.api.WebSocketPolicy;
37  import org.eclipse.jetty.websocket.api.extensions.Frame;
38  import org.eclipse.jetty.websocket.common.message.MessageInputStream;
39  import org.eclipse.jetty.websocket.common.message.MessageReader;
40  import org.eclipse.jetty.websocket.jsr356.JsrPongMessage;
41  import org.eclipse.jetty.websocket.jsr356.JsrSession;
42  import org.eclipse.jetty.websocket.jsr356.MessageHandlerWrapper;
43  import org.eclipse.jetty.websocket.jsr356.MessageType;
44  import org.eclipse.jetty.websocket.jsr356.messages.BinaryPartialMessage;
45  import org.eclipse.jetty.websocket.jsr356.messages.BinaryWholeMessage;
46  import org.eclipse.jetty.websocket.jsr356.messages.TextPartialMessage;
47  import org.eclipse.jetty.websocket.jsr356.messages.TextWholeMessage;
48  
49  /**
50   * EventDriver for websocket that extend from {@link javax.websocket.Endpoint}
51   */
52  public class JsrEndpointEventDriver extends AbstractJsrEventDriver
53  {
54      private static final Logger LOG = Log.getLogger(JsrEndpointEventDriver.class);
55  
56      private final Endpoint endpoint;
57      private Map<String, String> pathParameters;
58  
59      public JsrEndpointEventDriver(WebSocketPolicy policy, EndpointInstance endpointInstance)
60      {
61          super(policy,endpointInstance);
62          this.endpoint = (Endpoint)endpointInstance.getEndpoint();
63      }
64  
65      @Override
66      public void init(JsrSession jsrsession)
67      {
68          jsrsession.setPathParameters(pathParameters);
69      }
70  
71      @Override
72      public void onBinaryFrame(ByteBuffer buffer, boolean fin) throws IOException
73      {
74          if (activeMessage == null)
75          {
76              final MessageHandlerWrapper wrapper = jsrsession.getMessageHandlerWrapper(MessageType.BINARY);
77              if (wrapper == null)
78              {
79                  if (LOG.isDebugEnabled())
80                  {
81                      LOG.debug("No BINARY MessageHandler declared");
82                  }
83                  return;
84              }
85              if (wrapper.wantsPartialMessages())
86              {
87                  activeMessage = new BinaryPartialMessage(wrapper);
88              }
89              else if (wrapper.wantsStreams())
90              {
91                  final MessageInputStream stream = new MessageInputStream();
92                  activeMessage = stream;
93                  dispatch(new Runnable()
94                  {
95                      @SuppressWarnings("unchecked")
96                      @Override
97                      public void run()
98                      {
99                          MessageHandler.Whole<InputStream> handler = (Whole<InputStream>)wrapper.getHandler();
100                         handler.onMessage(stream);
101                     }
102                 });
103             }
104             else
105             {
106                 activeMessage = new BinaryWholeMessage(this,wrapper);
107             }
108         }
109 
110         activeMessage.appendFrame(buffer,fin);
111 
112         if (fin)
113         {
114             activeMessage.messageComplete();
115             activeMessage = null;
116         }
117     }
118 
119     @Override
120     public void onBinaryMessage(byte[] data)
121     {
122         /* Ignored, handled by BinaryWholeMessage */
123     }
124 
125     @Override
126     protected void onClose(CloseReason closereason)
127     {
128         endpoint.onClose(this.jsrsession,closereason);
129     }
130 
131     @Override
132     public void onConnect()
133     {
134         if (LOG.isDebugEnabled())
135         {
136             LOG.debug("onConnect({}, {})",jsrsession,config);
137         }
138 
139         // Let unhandled exceptions flow out
140         endpoint.onOpen(jsrsession,config);
141     }
142 
143     @Override
144     public void onError(Throwable cause)
145     {
146         LOG.warn(cause);
147         try
148         {
149             endpoint.onError(jsrsession,cause);
150         }
151         catch (Throwable t)
152         {
153             LOG.warn("Unable to report to onError due to exception",t);
154         }
155     }
156 
157     @Override
158     public void onFrame(Frame frame)
159     {
160         /* Ignored, not supported by JSR-356 */
161     }
162 
163     @Override
164     public void onInputStream(InputStream stream)
165     {
166         /* Ignored, handled by BinaryStreamMessage */
167     }
168 
169     @Override
170     public void onReader(Reader reader)
171     {
172         /* Ignored, handled by TextStreamMessage */
173     }
174 
175     @Override
176     public void onTextFrame(ByteBuffer buffer, boolean fin) throws IOException
177     {
178         if (activeMessage == null)
179         {
180             final MessageHandlerWrapper wrapper = jsrsession.getMessageHandlerWrapper(MessageType.TEXT);
181             if (wrapper == null)
182             {
183                 if (LOG.isDebugEnabled())
184                 {
185                     LOG.debug("No TEXT MessageHandler declared");
186                 }
187                 return;
188             }
189             if (wrapper.wantsPartialMessages())
190             {
191                 activeMessage = new TextPartialMessage(wrapper);
192             }
193             else if (wrapper.wantsStreams())
194             {
195                 final MessageReader stream = new MessageReader(new MessageInputStream());
196                 activeMessage = stream;
197 
198                 dispatch(new Runnable()
199                 {
200                     @SuppressWarnings("unchecked")
201                     @Override
202                     public void run()
203                     {
204                         MessageHandler.Whole<Reader> handler = (Whole<Reader>)wrapper.getHandler();
205                         handler.onMessage(stream);
206                     }
207                 });
208             }
209             else
210             {
211                 activeMessage = new TextWholeMessage(this,wrapper);
212             }
213         }
214 
215         activeMessage.appendFrame(buffer,fin);
216 
217         if (fin)
218         {
219             activeMessage.messageComplete();
220             activeMessage = null;
221         }
222     }
223 
224     @Override
225     public void onTextMessage(String message)
226     {
227         /* Ignored, handled by TextWholeMessage */
228     }
229     
230     @Override
231     public void onPing(ByteBuffer buffer)
232     {
233         onPongMessage(buffer);
234     }
235 
236     @Override
237     public void onPong(ByteBuffer buffer)
238     {
239         onPongMessage(buffer);
240     }
241 
242     private void onPongMessage(ByteBuffer buffer)
243     {
244         final MessageHandlerWrapper wrapper = jsrsession.getMessageHandlerWrapper(MessageType.PONG);
245         if (wrapper == null)
246         {
247             if (LOG.isDebugEnabled())
248             {
249                 LOG.debug("No PONG MessageHandler declared");
250             }
251             return;
252         }
253         
254         ByteBuffer pongBuf = null;
255         
256         if (BufferUtil.isEmpty(buffer))
257         {
258             pongBuf = BufferUtil.EMPTY_BUFFER;
259         }
260         else
261         {
262             pongBuf = ByteBuffer.allocate(buffer.remaining());
263             BufferUtil.put(buffer,pongBuf);
264             BufferUtil.flipToFlush(pongBuf,0);
265         }
266 
267         @SuppressWarnings("unchecked")
268         Whole<PongMessage> pongHandler = (Whole<PongMessage>)wrapper.getHandler();
269         pongHandler.onMessage(new JsrPongMessage(pongBuf));
270     }
271     
272     @Override
273     public void setPathParameters(Map<String, String> pathParameters)
274     {
275         this.pathParameters = pathParameters;
276     }
277 
278     @Override
279     public String toString()
280     {
281         return String.format("%s[%s]",JsrEndpointEventDriver.class.getSimpleName(),endpoint.getClass().getName());
282     }
283 }