1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.websocket.common.extensions;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.List;
24 import java.util.ListIterator;
25 import java.util.Queue;
26
27 import org.eclipse.jetty.util.ConcurrentArrayQueue;
28 import org.eclipse.jetty.util.IteratingCallback;
29 import org.eclipse.jetty.util.annotation.ManagedAttribute;
30 import org.eclipse.jetty.util.annotation.ManagedObject;
31 import org.eclipse.jetty.util.component.ContainerLifeCycle;
32 import org.eclipse.jetty.util.log.Log;
33 import org.eclipse.jetty.util.log.Logger;
34 import org.eclipse.jetty.websocket.api.BatchMode;
35 import org.eclipse.jetty.websocket.api.WebSocketPolicy;
36 import org.eclipse.jetty.websocket.api.WriteCallback;
37 import org.eclipse.jetty.websocket.api.extensions.Extension;
38 import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
39 import org.eclipse.jetty.websocket.api.extensions.ExtensionFactory;
40 import org.eclipse.jetty.websocket.api.extensions.Frame;
41 import org.eclipse.jetty.websocket.api.extensions.IncomingFrames;
42 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
43 import org.eclipse.jetty.websocket.common.Generator;
44 import org.eclipse.jetty.websocket.common.Parser;
45
46
47
48
49 @ManagedObject("Extension Stack")
50 public class ExtensionStack extends ContainerLifeCycle implements IncomingFrames, OutgoingFrames
51 {
52 private static final Logger LOG = Log.getLogger(ExtensionStack.class);
53
54 private final Queue<FrameEntry> entries = new ConcurrentArrayQueue<>();
55 private final IteratingCallback flusher = new Flusher();
56 private final ExtensionFactory factory;
57 private List<Extension> extensions;
58 private IncomingFrames nextIncoming;
59 private OutgoingFrames nextOutgoing;
60
61 public ExtensionStack(ExtensionFactory factory)
62 {
63 this.factory = factory;
64 }
65
66 public void configure(Generator generator)
67 {
68 generator.configureFromExtensions(extensions);
69 }
70
71 public void configure(Parser parser)
72 {
73 parser.configureFromExtensions(extensions);
74 }
75
76 @Override
77 protected void doStart() throws Exception
78 {
79 super.doStart();
80 LOG.debug("doStart");
81
82
83 if ((extensions != null) && (extensions.size() > 0))
84 {
85 ListIterator<Extension> exts = extensions.listIterator();
86
87
88 while (exts.hasNext())
89 {
90 Extension ext = exts.next();
91 ext.setNextOutgoingFrames(nextOutgoing);
92 nextOutgoing = ext;
93 }
94
95
96 while (exts.hasPrevious())
97 {
98 Extension ext = exts.previous();
99 ext.setNextIncomingFrames(nextIncoming);
100 nextIncoming = ext;
101 }
102 }
103 }
104
105 @Override
106 public void dump(Appendable out, String indent) throws IOException
107 {
108 super.dump(out,indent);
109
110 IncomingFrames websocket = getLastIncoming();
111 OutgoingFrames network = getLastOutgoing();
112
113 out.append(indent).append(" +- Stack").append(System.lineSeparator());
114 out.append(indent).append(" +- Network : ").append(network.toString()).append(System.lineSeparator());
115 for (Extension ext : extensions)
116 {
117 out.append(indent).append(" +- Extension: ").append(ext.toString()).append(System.lineSeparator());
118 }
119 out.append(indent).append(" +- Websocket: ").append(websocket.toString()).append(System.lineSeparator());
120 }
121
122 @ManagedAttribute(name = "Extension List", readonly = true)
123 public List<Extension> getExtensions()
124 {
125 return extensions;
126 }
127
128 private IncomingFrames getLastIncoming()
129 {
130 IncomingFrames last = nextIncoming;
131 boolean done = false;
132 while (!done)
133 {
134 if (last instanceof AbstractExtension)
135 {
136 last = ((AbstractExtension)last).getNextIncoming();
137 }
138 else
139 {
140 done = true;
141 }
142 }
143 return last;
144 }
145
146 private OutgoingFrames getLastOutgoing()
147 {
148 OutgoingFrames last = nextOutgoing;
149 boolean done = false;
150 while (!done)
151 {
152 if (last instanceof AbstractExtension)
153 {
154 last = ((AbstractExtension)last).getNextOutgoing();
155 }
156 else
157 {
158 done = true;
159 }
160 }
161 return last;
162 }
163
164
165
166
167
168
169 public List<ExtensionConfig> getNegotiatedExtensions()
170 {
171 List<ExtensionConfig> ret = new ArrayList<>();
172 if (extensions == null)
173 {
174 return ret;
175 }
176
177 for (Extension ext : extensions)
178 {
179 if (ext.getName().charAt(0) == '@')
180 {
181
182 continue;
183 }
184 ret.add(ext.getConfig());
185 }
186 return ret;
187 }
188
189 @ManagedAttribute(name = "Next Incoming Frames Handler", readonly = true)
190 public IncomingFrames getNextIncoming()
191 {
192 return nextIncoming;
193 }
194
195 @ManagedAttribute(name = "Next Outgoing Frames Handler", readonly = true)
196 public OutgoingFrames getNextOutgoing()
197 {
198 return nextOutgoing;
199 }
200
201 public boolean hasNegotiatedExtensions()
202 {
203 return (this.extensions != null) && (this.extensions.size() > 0);
204 }
205
206 @Override
207 public void incomingError(Throwable e)
208 {
209 nextIncoming.incomingError(e);
210 }
211
212 @Override
213 public void incomingFrame(Frame frame)
214 {
215 nextIncoming.incomingFrame(frame);
216 }
217
218
219
220
221
222
223
224
225
226 public void negotiate(List<ExtensionConfig> configs)
227 {
228 LOG.debug("Extension Configs={}",configs);
229 this.extensions = new ArrayList<>();
230
231 String rsvClaims[] = new String[3];
232
233 for (ExtensionConfig config : configs)
234 {
235 Extension ext = factory.newInstance(config);
236 if (ext == null)
237 {
238
239 continue;
240 }
241
242
243 if (ext.isRsv1User() && (rsvClaims[0] != null))
244 {
245 LOG.debug("Not adding extension {}. Extension {} already claimed RSV1",config,rsvClaims[0]);
246 continue;
247 }
248 if (ext.isRsv2User() && (rsvClaims[1] != null))
249 {
250 LOG.debug("Not adding extension {}. Extension {} already claimed RSV2",config,rsvClaims[1]);
251 continue;
252 }
253 if (ext.isRsv3User() && (rsvClaims[2] != null))
254 {
255 LOG.debug("Not adding extension {}. Extension {} already claimed RSV3",config,rsvClaims[2]);
256 continue;
257 }
258
259
260 extensions.add(ext);
261 addBean(ext);
262
263 LOG.debug("Adding Extension: {}",config);
264
265
266 if (ext.isRsv1User())
267 {
268 rsvClaims[0] = ext.getName();
269 }
270 if (ext.isRsv2User())
271 {
272 rsvClaims[1] = ext.getName();
273 }
274 if (ext.isRsv3User())
275 {
276 rsvClaims[2] = ext.getName();
277 }
278 }
279 }
280
281 @Override
282 public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
283 {
284 FrameEntry entry = new FrameEntry(frame,callback,batchMode);
285 LOG.debug("Queuing {}",entry);
286 entries.offer(entry);
287 flusher.iterate();
288 }
289
290 public void setNextIncoming(IncomingFrames nextIncoming)
291 {
292 this.nextIncoming = nextIncoming;
293 }
294
295 public void setNextOutgoing(OutgoingFrames nextOutgoing)
296 {
297 this.nextOutgoing = nextOutgoing;
298 }
299
300 public void setPolicy(WebSocketPolicy policy)
301 {
302 for (Extension extension : extensions)
303 {
304 if (extension instanceof AbstractExtension)
305 {
306 ((AbstractExtension)extension).setPolicy(policy);
307 }
308 }
309 }
310
311 @Override
312 public String toString()
313 {
314 StringBuilder s = new StringBuilder();
315 s.append("ExtensionStack[");
316 s.append("queueSize=").append(entries.size());
317 s.append(",extensions=");
318 if (extensions == null)
319 {
320 s.append("<null>");
321 }
322 else
323 {
324 s.append('[');
325 boolean delim = false;
326 for (Extension ext : extensions)
327 {
328 if (delim)
329 {
330 s.append(',');
331 }
332 if (ext == null)
333 {
334 s.append("<null>");
335 }
336 else
337 {
338 s.append(ext.getName());
339 }
340 delim = true;
341 }
342 s.append(']');
343 }
344 s.append(",incoming=").append((this.nextIncoming == null)?"<null>":this.nextIncoming.getClass().getName());
345 s.append(",outgoing=").append((this.nextOutgoing == null)?"<null>":this.nextOutgoing.getClass().getName());
346 s.append("]");
347 return s.toString();
348 }
349
350 private static class FrameEntry
351 {
352 private final Frame frame;
353 private final WriteCallback callback;
354 private final BatchMode batchMode;
355
356 private FrameEntry(Frame frame, WriteCallback callback, BatchMode batchMode)
357 {
358 this.frame = frame;
359 this.callback = callback;
360 this.batchMode = batchMode;
361 }
362
363 @Override
364 public String toString()
365 {
366 return frame.toString();
367 }
368 }
369
370 private class Flusher extends IteratingCallback implements WriteCallback
371 {
372 private FrameEntry current;
373
374 @Override
375 protected Action process() throws Exception
376 {
377 current = entries.poll();
378 if (current == null)
379 {
380 LOG.debug("Entering IDLE");
381 return Action.IDLE;
382 }
383 LOG.debug("Processing {}",current);
384 nextOutgoing.outgoingFrame(current.frame,this,current.batchMode);
385 return Action.SCHEDULED;
386 }
387
388 @Override
389 protected void completed()
390 {
391
392 }
393
394 @Override
395 public void writeSuccess()
396 {
397
398
399 notifyCallbackSuccess(current.callback);
400 succeeded();
401 }
402
403 @Override
404 public void writeFailed(Throwable x)
405 {
406
407
408
409
410
411 notifyCallbackFailure(current.callback,x);
412 succeeded();
413 }
414
415 private void notifyCallbackSuccess(WriteCallback callback)
416 {
417 try
418 {
419 if (callback != null)
420 callback.writeSuccess();
421 }
422 catch (Throwable x)
423 {
424 LOG.debug("Exception while notifying success of callback " + callback,x);
425 }
426 }
427
428 private void notifyCallbackFailure(WriteCallback callback, Throwable failure)
429 {
430 try
431 {
432 if (callback != null)
433 callback.writeFailed(failure);
434 }
435 catch (Throwable x)
436 {
437 LOG.debug("Exception while notifying failure of callback " + callback,x);
438 }
439 }
440 }
441 }