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.message;
20
21 import java.io.IOException;
22 import java.io.Writer;
23 import java.nio.ByteBuffer;
24
25 import org.eclipse.jetty.io.ByteBufferPool;
26 import org.eclipse.jetty.util.BufferUtil;
27 import org.eclipse.jetty.util.log.Log;
28 import org.eclipse.jetty.util.log.Logger;
29 import org.eclipse.jetty.websocket.api.WriteCallback;
30 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
31 import org.eclipse.jetty.websocket.common.BlockingWriteCallback;
32 import org.eclipse.jetty.websocket.common.WebSocketSession;
33 import org.eclipse.jetty.websocket.common.frames.TextFrame;
34
35
36
37
38
39
40 public class MessageWriter extends Writer
41 {
42 private static final Logger LOG = Log.getLogger(MessageWriter.class);
43 private final OutgoingFrames outgoing;
44 private final ByteBufferPool bufferPool;
45 private final BlockingWriteCallback blocker;
46 private long frameCount = 0;
47 private TextFrame frame;
48 private ByteBuffer buffer;
49 private Utf8CharBuffer utf;
50 private WriteCallback callback;
51 private boolean closed = false;
52
53 public MessageWriter(OutgoingFrames outgoing, int bufferSize, ByteBufferPool bufferPool)
54 {
55 this.outgoing = outgoing;
56 this.bufferPool = bufferPool;
57 this.blocker = new BlockingWriteCallback();
58 this.buffer = bufferPool.acquire(bufferSize,true);
59 BufferUtil.flipToFill(buffer);
60 this.utf = Utf8CharBuffer.wrap(buffer);
61 this.frame = new TextFrame();
62 }
63
64 public MessageWriter(WebSocketSession session)
65 {
66 this(session.getOutgoingHandler(),session.getPolicy().getMaxTextMessageBufferSize(),session.getBufferPool());
67 }
68
69 private void assertNotClosed() throws IOException
70 {
71 if (closed)
72 {
73 IOException e = new IOException("Stream is closed");
74 notifyFailure(e);
75 throw e;
76 }
77 }
78
79 @Override
80 public synchronized void close() throws IOException
81 {
82 assertNotClosed();
83
84
85 flush(true);
86
87
88 closed = true;
89 if (callback != null)
90 {
91 callback.writeSuccess();
92 }
93 bufferPool.release(buffer);
94 LOG.debug("closed (frame count={})",frameCount);
95 }
96
97 @Override
98 public void flush() throws IOException
99 {
100 assertNotClosed();
101
102
103 flush(false);
104 }
105
106
107
108
109
110
111
112
113 private synchronized void flush(boolean fin) throws IOException
114 {
115 ByteBuffer data = utf.getByteBuffer();
116 frame.setPayload(data);
117 frame.setFin(fin);
118
119 try
120 {
121 outgoing.outgoingFrame(frame,blocker);
122
123 blocker.block();
124
125
126 utf.clear();
127 frameCount++;
128 frame.setIsContinuation();
129 }
130 catch (IOException e)
131 {
132 notifyFailure(e);
133 throw e;
134 }
135 }
136
137 private void notifyFailure(IOException e)
138 {
139 if (callback != null)
140 {
141 callback.writeFailed(e);
142 }
143 }
144
145 public void setCallback(WriteCallback callback)
146 {
147 this.callback = callback;
148 }
149
150 @Override
151 public void write(char[] cbuf) throws IOException
152 {
153 try
154 {
155 this.write(cbuf,0,cbuf.length);
156 }
157 catch (IOException e)
158 {
159 notifyFailure(e);
160 throw e;
161 }
162 }
163
164 @Override
165 public void write(char[] cbuf, int off, int len) throws IOException
166 {
167 assertNotClosed();
168 int left = len;
169 int offset = off;
170 while (left > 0)
171 {
172 int space = utf.remaining();
173 int size = Math.min(space,left);
174 assert (space > 0);
175 assert (size > 0);
176 utf.append(cbuf,offset,size);
177 left -= size;
178 if (left > 0)
179 {
180 flush(false);
181 }
182 offset += size;
183 }
184 }
185
186 @Override
187 public void write(int c) throws IOException
188 {
189 assertNotClosed();
190
191
192 utf.append(c);
193 if (utf.remaining() <= 0)
194 {
195 flush(false);
196 }
197 }
198 }