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;
20
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.nio.ByteBuffer;
24 import java.nio.charset.StandardCharsets;
25 import java.util.concurrent.Future;
26 import java.util.concurrent.atomic.AtomicInteger;
27
28 import org.eclipse.jetty.util.BufferUtil;
29 import org.eclipse.jetty.util.log.Log;
30 import org.eclipse.jetty.util.log.Logger;
31 import org.eclipse.jetty.websocket.api.BatchMode;
32 import org.eclipse.jetty.websocket.api.RemoteEndpoint;
33 import org.eclipse.jetty.websocket.api.WriteCallback;
34 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
35 import org.eclipse.jetty.websocket.common.BlockingWriteCallback.WriteBlocker;
36 import org.eclipse.jetty.websocket.common.frames.BinaryFrame;
37 import org.eclipse.jetty.websocket.common.frames.ContinuationFrame;
38 import org.eclipse.jetty.websocket.common.frames.DataFrame;
39 import org.eclipse.jetty.websocket.common.frames.PingFrame;
40 import org.eclipse.jetty.websocket.common.frames.PongFrame;
41 import org.eclipse.jetty.websocket.common.frames.TextFrame;
42 import org.eclipse.jetty.websocket.common.io.FrameFlusher;
43 import org.eclipse.jetty.websocket.common.io.FutureWriteCallback;
44
45
46
47
48 public class WebSocketRemoteEndpoint implements RemoteEndpoint
49 {
50 private enum MsgType
51 {
52 BLOCKING,
53 ASYNC,
54 STREAMING,
55 PARTIAL_TEXT,
56 PARTIAL_BINARY
57 }
58
59 private static final WriteCallback NOOP_CALLBACK = new WriteCallback()
60 {
61 @Override
62 public void writeSuccess()
63 {
64 }
65
66 @Override
67 public void writeFailed(Throwable x)
68 {
69 }
70 };
71
72 private static final Logger LOG = Log.getLogger(WebSocketRemoteEndpoint.class);
73
74 private final static int ASYNC_MASK = 0x0000FFFF;
75 private final static int BLOCK_MASK = 0x00010000;
76 private final static int STREAM_MASK = 0x00020000;
77 private final static int PARTIAL_TEXT_MASK = 0x00040000;
78 private final static int PARTIAL_BINARY_MASK = 0x00080000;
79
80 private final LogicalConnection connection;
81 private final OutgoingFrames outgoing;
82 private final AtomicInteger msgState = new AtomicInteger();
83 private final BlockingWriteCallback blocker = new BlockingWriteCallback();
84 private volatile BatchMode batchMode;
85
86 public WebSocketRemoteEndpoint(LogicalConnection connection, OutgoingFrames outgoing)
87 {
88 this(connection, outgoing, BatchMode.AUTO);
89 }
90
91 public WebSocketRemoteEndpoint(LogicalConnection connection, OutgoingFrames outgoing, BatchMode batchMode)
92 {
93 if (connection == null)
94 {
95 throw new IllegalArgumentException("LogicalConnection cannot be null");
96 }
97 this.connection = connection;
98 this.outgoing = outgoing;
99 this.batchMode = batchMode;
100 }
101
102 private void blockingWrite(WebSocketFrame frame) throws IOException
103 {
104 try(WriteBlocker b=blocker.acquireWriteBlocker())
105 {
106 uncheckedSendFrame(frame, b);
107 b.block();
108 }
109 }
110
111 private boolean lockMsg(MsgType type)
112 {
113
114
115
116
117
118
119
120 while (true)
121 {
122 int state = msgState.get();
123
124 switch (type)
125 {
126 case BLOCKING:
127 if ((state & (PARTIAL_BINARY_MASK + PARTIAL_TEXT_MASK)) != 0)
128 throw new IllegalStateException(String.format("Partial message pending %x for %s", state, type));
129 if ((state & BLOCK_MASK) != 0)
130 throw new IllegalStateException(String.format("Blocking message pending %x for %s", state, type));
131 if (msgState.compareAndSet(state, state | BLOCK_MASK))
132 return state == 0;
133 break;
134
135 case ASYNC:
136 if ((state & (PARTIAL_BINARY_MASK + PARTIAL_TEXT_MASK)) != 0)
137 throw new IllegalStateException(String.format("Partial message pending %x for %s", state, type));
138 if ((state & ASYNC_MASK) == ASYNC_MASK)
139 throw new IllegalStateException(String.format("Too many async sends: %x", state));
140 if (msgState.compareAndSet(state, state + 1))
141 return state == 0;
142 break;
143
144 case STREAMING:
145 if ((state & (PARTIAL_BINARY_MASK + PARTIAL_TEXT_MASK)) != 0)
146 throw new IllegalStateException(String.format("Partial message pending %x for %s", state, type));
147 if ((state & STREAM_MASK) != 0)
148 throw new IllegalStateException(String.format("Already streaming %x for %s", state, type));
149 if (msgState.compareAndSet(state, state | STREAM_MASK))
150 return state == 0;
151 break;
152
153 case PARTIAL_BINARY:
154 if (state == PARTIAL_BINARY_MASK)
155 return false;
156 if (state == 0)
157 {
158 if (msgState.compareAndSet(0, state | PARTIAL_BINARY_MASK))
159 return true;
160 }
161 throw new IllegalStateException(String.format("Cannot send %s in state %x", type, state));
162
163 case PARTIAL_TEXT:
164 if (state == PARTIAL_TEXT_MASK)
165 return false;
166 if (state == 0)
167 {
168 if (msgState.compareAndSet(0, state | PARTIAL_TEXT_MASK))
169 return true;
170 }
171 throw new IllegalStateException(String.format("Cannot send %s in state %x", type, state));
172 }
173 }
174 }
175
176 private void unlockMsg(MsgType type)
177 {
178 while (true)
179 {
180 int state = msgState.get();
181
182 switch (type)
183 {
184 case BLOCKING:
185 if ((state & BLOCK_MASK) == 0)
186 throw new IllegalStateException(String.format("Not Blocking in state %x", state));
187 if (msgState.compareAndSet(state, state & ~BLOCK_MASK))
188 return;
189 break;
190
191 case ASYNC:
192 if ((state & ASYNC_MASK) == 0)
193 throw new IllegalStateException(String.format("Not Async in %x", state));
194 if (msgState.compareAndSet(state, state - 1))
195 return;
196 break;
197
198 case STREAMING:
199 if ((state & STREAM_MASK) == 0)
200 throw new IllegalStateException(String.format("Not Streaming in state %x", state));
201 if (msgState.compareAndSet(state, state & ~STREAM_MASK))
202 return;
203 break;
204
205 case PARTIAL_BINARY:
206 if (msgState.compareAndSet(PARTIAL_BINARY_MASK, 0))
207 return;
208 throw new IllegalStateException(String.format("Not Partial Binary in state %x", state));
209
210 case PARTIAL_TEXT:
211 if (msgState.compareAndSet(PARTIAL_TEXT_MASK, 0))
212 return;
213 throw new IllegalStateException(String.format("Not Partial Text in state %x", state));
214
215 }
216 }
217 }
218
219
220
221
222
223
224 public InetSocketAddress getInetSocketAddress()
225 {
226 if(connection == null)
227 return null;
228 return connection.getRemoteAddress();
229 }
230
231
232
233
234
235
236
237 private Future<Void> sendAsyncFrame(WebSocketFrame frame)
238 {
239 FutureWriteCallback future = new FutureWriteCallback();
240 uncheckedSendFrame(frame, future);
241 return future;
242 }
243
244
245
246
247 @Override
248 public void sendBytes(ByteBuffer data) throws IOException
249 {
250 lockMsg(MsgType.BLOCKING);
251 try
252 {
253 connection.getIOState().assertOutputOpen();
254 if (LOG.isDebugEnabled())
255 {
256 LOG.debug("sendBytes with {}", BufferUtil.toDetailString(data));
257 }
258 blockingWrite(new BinaryFrame().setPayload(data));
259 }
260 finally
261 {
262 unlockMsg(MsgType.BLOCKING);
263 }
264 }
265
266 @Override
267 public Future<Void> sendBytesByFuture(ByteBuffer data)
268 {
269 lockMsg(MsgType.ASYNC);
270 try
271 {
272 if (LOG.isDebugEnabled())
273 {
274 LOG.debug("sendBytesByFuture with {}", BufferUtil.toDetailString(data));
275 }
276 return sendAsyncFrame(new BinaryFrame().setPayload(data));
277 }
278 finally
279 {
280 unlockMsg(MsgType.ASYNC);
281 }
282 }
283
284 @Override
285 public void sendBytes(ByteBuffer data, WriteCallback callback)
286 {
287 lockMsg(MsgType.ASYNC);
288 try
289 {
290 if (LOG.isDebugEnabled())
291 {
292 LOG.debug("sendBytes({}, {})", BufferUtil.toDetailString(data), callback);
293 }
294 uncheckedSendFrame(new BinaryFrame().setPayload(data), callback == null ? NOOP_CALLBACK : callback);
295 }
296 finally
297 {
298 unlockMsg(MsgType.ASYNC);
299 }
300 }
301
302 public void uncheckedSendFrame(WebSocketFrame frame, WriteCallback callback)
303 {
304 try
305 {
306 BatchMode batchMode = BatchMode.OFF;
307 if (frame.isDataFrame())
308 batchMode = getBatchMode();
309 connection.getIOState().assertOutputOpen();
310 outgoing.outgoingFrame(frame, callback, batchMode);
311 }
312 catch (IOException e)
313 {
314 callback.writeFailed(e);
315 }
316 }
317
318 @Override
319 public void sendPartialBytes(ByteBuffer fragment, boolean isLast) throws IOException
320 {
321 boolean first = lockMsg(MsgType.PARTIAL_BINARY);
322 try
323 {
324 if (LOG.isDebugEnabled())
325 {
326 LOG.debug("sendPartialBytes({}, {})", BufferUtil.toDetailString(fragment), isLast);
327 }
328 DataFrame frame = first ? new BinaryFrame() : new ContinuationFrame();
329 frame.setPayload(fragment);
330 frame.setFin(isLast);
331 blockingWrite(frame);
332 }
333 finally
334 {
335 if (isLast)
336 unlockMsg(MsgType.PARTIAL_BINARY);
337 }
338 }
339
340 @Override
341 public void sendPartialString(String fragment, boolean isLast) throws IOException
342 {
343 boolean first = lockMsg(MsgType.PARTIAL_TEXT);
344 try
345 {
346 if (LOG.isDebugEnabled())
347 {
348 LOG.debug("sendPartialString({}, {})", fragment, isLast);
349 }
350 DataFrame frame = first ? new TextFrame() : new ContinuationFrame();
351 frame.setPayload(BufferUtil.toBuffer(fragment, StandardCharsets.UTF_8));
352 frame.setFin(isLast);
353 blockingWrite(frame);
354 }
355 finally
356 {
357 if (isLast)
358 unlockMsg(MsgType.PARTIAL_TEXT);
359 }
360 }
361
362 @Override
363 public void sendPing(ByteBuffer applicationData) throws IOException
364 {
365 if (LOG.isDebugEnabled())
366 {
367 LOG.debug("sendPing with {}", BufferUtil.toDetailString(applicationData));
368 }
369 sendAsyncFrame(new PingFrame().setPayload(applicationData));
370 }
371
372 @Override
373 public void sendPong(ByteBuffer applicationData) throws IOException
374 {
375 if (LOG.isDebugEnabled())
376 {
377 LOG.debug("sendPong with {}", BufferUtil.toDetailString(applicationData));
378 }
379 sendAsyncFrame(new PongFrame().setPayload(applicationData));
380 }
381
382 @Override
383 public void sendString(String text) throws IOException
384 {
385 lockMsg(MsgType.BLOCKING);
386 try
387 {
388 WebSocketFrame frame = new TextFrame().setPayload(text);
389 if (LOG.isDebugEnabled())
390 {
391 LOG.debug("sendString with {}", BufferUtil.toDetailString(frame.getPayload()));
392 }
393 blockingWrite(frame);
394 }
395 finally
396 {
397 unlockMsg(MsgType.BLOCKING);
398 }
399 }
400
401 @Override
402 public Future<Void> sendStringByFuture(String text)
403 {
404 lockMsg(MsgType.ASYNC);
405 try
406 {
407 TextFrame frame = new TextFrame().setPayload(text);
408 if (LOG.isDebugEnabled())
409 {
410 LOG.debug("sendStringByFuture with {}", BufferUtil.toDetailString(frame.getPayload()));
411 }
412 return sendAsyncFrame(frame);
413 }
414 finally
415 {
416 unlockMsg(MsgType.ASYNC);
417 }
418 }
419
420 @Override
421 public void sendString(String text, WriteCallback callback)
422 {
423 lockMsg(MsgType.ASYNC);
424 try
425 {
426 TextFrame frame = new TextFrame().setPayload(text);
427 if (LOG.isDebugEnabled())
428 {
429 LOG.debug("sendString({},{})", BufferUtil.toDetailString(frame.getPayload()), callback);
430 }
431 uncheckedSendFrame(frame, callback == null ? NOOP_CALLBACK : callback);
432 }
433 finally
434 {
435 unlockMsg(MsgType.ASYNC);
436 }
437 }
438
439 @Override
440 public BatchMode getBatchMode()
441 {
442 return batchMode;
443 }
444
445 @Override
446 public void setBatchMode(BatchMode batchMode)
447 {
448 this.batchMode = batchMode;
449 }
450
451 @Override
452 public void flush() throws IOException
453 {
454 lockMsg(MsgType.ASYNC);
455 try (WriteBlocker b = blocker.acquireWriteBlocker())
456 {
457 uncheckedSendFrame(FrameFlusher.FLUSH_FRAME, b);
458 b.block();
459 }
460 finally
461 {
462 unlockMsg(MsgType.ASYNC);
463 }
464 }
465
466 @Override
467 public String toString()
468 {
469 return String.format("%s@%x[batching=%b]", getClass().getSimpleName(), hashCode(), getBatchMode());
470 }
471 }