View Javadoc

1   //
2   //  ========================================================================
3   //  Copyright (c) 1995-2014 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.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   * Support for writing a single WebSocket TEXT message via a {@link Writer}
37   * <p>
38   * Note: Per WebSocket spec, all WebSocket TEXT messages must be encoded in UTF-8
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          // finish sending whatever in the buffer with FIN=true
85          flush(true);
86  
87          // close stream
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         // flush whatever is in the buffer with FIN=false
103         flush(false);
104     }
105 
106     /**
107      * Flush whatever is in the buffer.
108      * 
109      * @param fin
110      *            fin flag
111      * @throws IOException
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             // block on write
123             blocker.block();
124             // write success
125             // clear utf buffer
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; // bytes left to write
169         int offset = off; // offset within provided array
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); // append with utf logic
177             left -= size; // decrement char left
178             if (left > 0)
179             {
180                 flush(false);
181             }
182             offset += size; // increment offset
183         }
184     }
185 
186     @Override
187     public void write(int c) throws IOException
188     {
189         assertNotClosed();
190 
191         // buffer up to limit, flush once buffer reached.
192         utf.append(c); // append with utf logic
193         if (utf.remaining() <= 0)
194         {
195             flush(false);
196         }
197     }
198 }