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 ret.add(ext.getConfig());
180 }
181 return ret;
182 }
183
184 @ManagedAttribute(name = "Next Incoming Frames Handler", readonly = true)
185 public IncomingFrames getNextIncoming()
186 {
187 return nextIncoming;
188 }
189
190 @ManagedAttribute(name = "Next Outgoing Frames Handler", readonly = true)
191 public OutgoingFrames getNextOutgoing()
192 {
193 return nextOutgoing;
194 }
195
196 public boolean hasNegotiatedExtensions()
197 {
198 return (this.extensions != null) && (this.extensions.size() > 0);
199 }
200
201 @Override
202 public void incomingError(Throwable e)
203 {
204 nextIncoming.incomingError(e);
205 }
206
207 @Override
208 public void incomingFrame(Frame frame)
209 {
210 nextIncoming.incomingFrame(frame);
211 }
212
213
214
215
216
217
218
219
220
221 public void negotiate(List<ExtensionConfig> configs)
222 {
223 LOG.debug("Extension Configs={}",configs);
224 this.extensions = new ArrayList<>();
225
226 String rsvClaims[] = new String[3];
227
228 for (ExtensionConfig config : configs)
229 {
230 Extension ext = factory.newInstance(config);
231 if (ext == null)
232 {
233
234 continue;
235 }
236
237
238 if (ext.isRsv1User() && (rsvClaims[0] != null))
239 {
240 LOG.debug("Not adding extension {}. Extension {} already claimed RSV1",config,rsvClaims[0]);
241 continue;
242 }
243 if (ext.isRsv2User() && (rsvClaims[1] != null))
244 {
245 LOG.debug("Not adding extension {}. Extension {} already claimed RSV2",config,rsvClaims[1]);
246 continue;
247 }
248 if (ext.isRsv3User() && (rsvClaims[2] != null))
249 {
250 LOG.debug("Not adding extension {}. Extension {} already claimed RSV3",config,rsvClaims[2]);
251 continue;
252 }
253
254
255 extensions.add(ext);
256 addBean(ext);
257
258 LOG.debug("Adding Extension: {}",config);
259
260
261 if (ext.isRsv1User())
262 {
263 rsvClaims[0] = ext.getName();
264 }
265 if (ext.isRsv2User())
266 {
267 rsvClaims[1] = ext.getName();
268 }
269 if (ext.isRsv3User())
270 {
271 rsvClaims[2] = ext.getName();
272 }
273 }
274 }
275
276 @Override
277 public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
278 {
279 FrameEntry entry = new FrameEntry(frame, callback, batchMode);
280 LOG.debug("Queuing {}", entry);
281 entries.offer(entry);
282 flusher.iterate();
283 }
284
285 public void setNextIncoming(IncomingFrames nextIncoming)
286 {
287 this.nextIncoming = nextIncoming;
288 }
289
290 public void setNextOutgoing(OutgoingFrames nextOutgoing)
291 {
292 this.nextOutgoing = nextOutgoing;
293 }
294
295 public void setPolicy(WebSocketPolicy policy)
296 {
297 for (Extension extension : extensions)
298 {
299 if (extension instanceof AbstractExtension)
300 {
301 ((AbstractExtension)extension).setPolicy(policy);
302 }
303 }
304 }
305
306 @Override
307 public String toString()
308 {
309 StringBuilder s = new StringBuilder();
310 s.append("ExtensionStack[");
311 s.append("queueSize=").append(entries.size());
312 s.append(",extensions=");
313 if (extensions == null)
314 {
315 s.append("<null>");
316 }
317 else
318 {
319 s.append('[');
320 boolean delim = false;
321 for (Extension ext : extensions)
322 {
323 if (delim)
324 {
325 s.append(',');
326 }
327 if (ext == null)
328 {
329 s.append("<null>");
330 }
331 else
332 {
333 s.append(ext.getName());
334 }
335 delim = true;
336 }
337 s.append(']');
338 }
339 s.append(",incoming=").append((this.nextIncoming == null)?"<null>":this.nextIncoming.getClass().getName());
340 s.append(",outgoing=").append((this.nextOutgoing == null)?"<null>":this.nextOutgoing.getClass().getName());
341 s.append("]");
342 return s.toString();
343 }
344
345 private static class FrameEntry
346 {
347 private final Frame frame;
348 private final WriteCallback callback;
349 private final BatchMode batchMode;
350
351 private FrameEntry(Frame frame, WriteCallback callback, BatchMode batchMode)
352 {
353 this.frame = frame;
354 this.callback = callback;
355 this.batchMode = batchMode;
356 }
357
358 @Override
359 public String toString()
360 {
361 return frame.toString();
362 }
363 }
364
365 private class Flusher extends IteratingCallback implements WriteCallback
366 {
367 private FrameEntry current;
368
369 @Override
370 protected Action process() throws Exception
371 {
372 current = entries.poll();
373 if (current == null)
374 {
375 LOG.debug("Entering IDLE");
376 return Action.IDLE;
377 }
378 LOG.debug("Processing {}",current);
379 nextOutgoing.outgoingFrame(current.frame,this,current.batchMode);
380 return Action.SCHEDULED;
381 }
382
383 @Override
384 protected void completed()
385 {
386
387 }
388
389 @Override
390 public void writeSuccess()
391 {
392
393
394 notifyCallbackSuccess(current.callback);
395 succeeded();
396 }
397
398 @Override
399 public void writeFailed(Throwable x)
400 {
401
402
403
404
405
406 notifyCallbackFailure(current.callback, x);
407 succeeded();
408 }
409
410 private void notifyCallbackSuccess(WriteCallback callback)
411 {
412 try
413 {
414 if (callback != null)
415 callback.writeSuccess();
416 }
417 catch (Throwable x)
418 {
419 LOG.debug("Exception while notifying success of callback " + callback, x);
420 }
421 }
422
423 private void notifyCallbackFailure(WriteCallback callback, Throwable failure)
424 {
425 try
426 {
427 if (callback != null)
428 callback.writeFailed(failure);
429 }
430 catch (Throwable x)
431 {
432 LOG.debug("Exception while notifying failure of callback " + callback, x);
433 }
434 }
435 }
436 }