Hooray... sonews/0.5.0 final
HG: Enter commit message. Lines beginning with 'HG:' are removed.
HG: Remove all lines to abort the collapse operation.
3 * see AUTHORS for the list of contributors
5 * This program is free software: you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation, either version 3 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>.
19 package org.sonews.daemon;
21 import java.nio.ByteBuffer;
22 import java.nio.channels.ClosedChannelException;
23 import java.util.ArrayList;
24 import java.util.List;
27 * Class holding ByteBuffers for SocketChannels/NNTPConnection.
28 * Due to the complex nature of AIO/NIO we must properly handle the line
29 * buffers for the input and output of the SocketChannels.
30 * @author Christian Lins
33 public class ChannelLineBuffers
37 * Size of one small buffer;
38 * per default this is 512 bytes to fit one standard line.
40 public static final int BUFFER_SIZE = 512;
42 private static int maxCachedBuffers = 2048; // Cached buffers maximum
44 private static final List<ByteBuffer> freeSmallBuffers
45 = new ArrayList<ByteBuffer>(maxCachedBuffers);
48 * Allocates a predefined number of direct ByteBuffers (allocated via
49 * ByteBuffer.allocateDirect()). This method is Thread-safe, but should only
52 public static void allocateDirect()
54 synchronized(freeSmallBuffers)
56 for(int n = 0; n < maxCachedBuffers; n++)
58 ByteBuffer buffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
59 freeSmallBuffers.add(buffer);
64 private ByteBuffer inputBuffer = newLineBuffer();
65 private List<ByteBuffer> outputBuffers = new ArrayList<ByteBuffer>();
68 * Add the given ByteBuffer to the list of buffers to be send to the client.
69 * This method is Thread-safe.
71 * @throws java.nio.channels.ClosedChannelException If the client channel was
74 public void addOutputBuffer(ByteBuffer buffer)
75 throws ClosedChannelException
77 if(outputBuffers == null)
79 throw new ClosedChannelException();
82 synchronized(outputBuffers)
84 outputBuffers.add(buffer);
89 * Currently a channel has only one input buffer. This *may* be a bottleneck
90 * and should investigated in the future.
92 * @return The input buffer associated with given channel.
94 public ByteBuffer getInputBuffer()
100 * Returns the current output buffer for writing(!) to SocketChannel.
102 * @return The next input buffer that contains unprocessed data or null
103 * if the connection was closed or there are no more unprocessed buffers.
105 public ByteBuffer getOutputBuffer()
107 synchronized(outputBuffers)
109 if(outputBuffers == null || outputBuffers.isEmpty())
115 ByteBuffer buffer = outputBuffers.get(0);
116 if(buffer.remaining() == 0)
118 outputBuffers.remove(0);
119 // Add old buffers to the list of free buffers
120 recycleBuffer(buffer);
121 buffer = getOutputBuffer();
129 * Goes through the input buffer of the given channel and searches
130 * for next line terminator. If a '\n' is found, the bytes up to the
131 * line terminator are returned as array of bytes (the line terminator
132 * is omitted). If none is found the method returns null.
134 * @return A ByteBuffer wrapping the line.
136 ByteBuffer nextInputLine()
138 if(inputBuffer == null)
143 synchronized(inputBuffer)
145 ByteBuffer buffer = inputBuffer;
147 // Mark the current write position
148 int mark = buffer.position();
150 // Set position to 0 and limit to current position
153 ByteBuffer lineBuffer = newLineBuffer();
155 while (buffer.position() < buffer.limit())
157 byte b = buffer.get();
160 // The bytes between the buffer's current position and its limit,
161 // if any, are copied to the beginning of the buffer. That is, the
162 // byte at index p = position() is copied to index zero, the byte at
163 // index p + 1 is copied to index one, and so forth until the byte
164 // at index limit() - 1 is copied to index n = limit() - 1 - p.
165 // The buffer's position is then set to n+1 and its limit is set to
169 lineBuffer.flip(); // limit to position, position to 0
178 buffer.limit(BUFFER_SIZE);
179 buffer.position(mark);
181 if(buffer.hasRemaining())
187 // In the first 512 was no newline found, so the input is not standard
188 // compliant. We return the current buffer as new line and add a space
189 // to the beginning of the next line which corrects some overlong header
191 inputBuffer = newLineBuffer();
192 inputBuffer.put((byte)' ');
200 * Returns a at least 512 bytes long ByteBuffer ready for usage.
201 * The method first try to reuse an already allocated (cached) buffer but
202 * if that fails returns a newly allocated direct buffer.
203 * Use recycleBuffer() method when you do not longer use the allocated buffer.
205 static ByteBuffer newLineBuffer()
207 ByteBuffer buf = null;
208 synchronized(freeSmallBuffers)
210 if(!freeSmallBuffers.isEmpty())
212 buf = freeSmallBuffers.remove(0);
218 // Allocate a non-direct buffer
219 buf = ByteBuffer.allocate(BUFFER_SIZE);
222 assert buf.position() == 0;
223 assert buf.limit() >= BUFFER_SIZE;
229 * Adds the given buffer to the list of free buffers if it is a valuable
230 * direct allocated buffer.
233 public static void recycleBuffer(ByteBuffer buffer)
235 assert buffer != null;
236 assert buffer.capacity() >= BUFFER_SIZE;
238 if(buffer.isDirect())
240 // Add old buffers to the list of free buffers
241 synchronized(freeSmallBuffers)
243 buffer.clear(); // Set position to 0 and limit to capacity
244 freeSmallBuffers.add(buffer);
246 } // if(buffer.isDirect())
250 * Recycles all buffers of this ChannelLineBuffers object.
252 public void recycleBuffers()
254 synchronized(inputBuffer)
256 recycleBuffer(inputBuffer);
257 this.inputBuffer = null;
260 synchronized(outputBuffers)
262 for(ByteBuffer buf : outputBuffers)
266 outputBuffers = null;