001: /*
002: * This file is part of the QuickServer library
003: * Copyright (C) QuickServer.org
004: *
005: * Use, modification, copying and distribution of this software is subject to
006: * the terms and conditions of the GNU Lesser General Public License.
007: * You should have received a copy of the GNU LGP License along with this
008: * library; if not, you can download a copy from <http://www.quickserver.org/>.
009: *
010: * For questions, suggestions, bug-reports, enhancement-requests etc.
011: * visit http://www.quickserver.org
012: *
013: */
014:
015: package org.quickserver.util.io;
016:
017: import java.io.*;
018: import java.nio.*;
019: import java.util.*;
020: import org.apache.commons.pool.ObjectPool;
021: import org.quickserver.net.server.ClientHandler;
022: import org.quickserver.net.server.impl.NonBlockingClientHandler;
023: import java.util.logging.*;
024:
025: /**
026: * This is an OutputStream constructed from list of ByteBuffers. This is
027: * used in non-blocking mode.
028: * @since 1.4.5
029: * @author Akshathkumar Shetty
030: */
031: public class ByteBufferOutputStream extends OutputStream {
032: private static Logger logger = Logger
033: .getLogger(ByteBufferOutputStream.class.getName());
034: static {
035: logger.setLevel(Level.INFO);
036: }
037:
038: /**
039: * Sets the debug flag. When debug is set to <code>true</code>
040: * one can see number of bytes written.
041: */
042: public static void setDebug(boolean flag) {
043: if (flag)
044: logger.setLevel(Level.FINEST);
045: else
046: logger.setLevel(Level.INFO);
047: }
048:
049: /**
050: * @since 1.4.7
051: */
052: public static boolean isLoggable(Level level) {
053: return logger.isLoggable(level);
054: }
055:
056: private ArrayList bufferList;
057: private ByteBuffer lastByteBuffer = null;
058: private NonBlockingClientHandler handler;
059: private Object toNotify = null;
060:
061: /**
062: * Creates a new ByteBufferOutputStream using the given list as its base
063: * and ClientHandler as the target channel.
064: */
065: public ByteBufferOutputStream(ArrayList bufferList,
066: ClientHandler handler) {
067: if (bufferList == null || handler == null)
068: throw new IllegalArgumentException(
069: "ArrayList or ClientHandler was null.");
070: this .bufferList = bufferList;
071: this .handler = (NonBlockingClientHandler) handler;
072: }
073:
074: public synchronized void close() {
075: if (lastByteBuffer != null) {
076: returnBufferBack(lastByteBuffer);
077: }
078: }
079:
080: public void flush() throws IOException {
081: if (bufferList.size() != 0 || lastByteBuffer != null) {
082: handler.registerWrite();
083: } else {
084: return;
085: }
086:
087: while (bufferList.size() >= 5) {
088: handler.waitTillFullyWritten();
089: }
090: }
091:
092: public synchronized void write(int b) throws IOException {
093: handler.isConnected();
094: ByteBuffer byteBuffer = null;
095: if (bufferList.size() != 0) {
096: byteBuffer = (ByteBuffer) bufferList.remove(bufferList
097: .size() - 1);
098: if (byteBuffer.remaining() == 0) {
099: bufferList.add(byteBuffer);
100: byteBuffer = null;
101: }
102: }
103: try {
104: if (byteBuffer == null) {
105: byteBuffer = (ByteBuffer) handler.getServer()
106: .getByteBufferPool().borrowObject();
107: }
108: } catch (Exception e) {
109: logger
110: .warning("Could not borrow ByteBufer from pool: "
111: + e);
112: throw new IOException(e.toString());
113: }
114: byteBuffer.put((byte) b);
115: bufferList.add(byteBuffer);
116: }
117:
118: public void write(byte[] b) throws IOException {
119: write(b, 0, b.length);
120: }
121:
122: public synchronized void write(byte[] b, int off, int len)
123: throws IOException {
124: if (len == 0) {
125: return;
126: }
127:
128: handler.isConnected();
129: ByteBuffer byteBuffer = null;
130: int remaining = 0;
131: int toWrite = len;
132:
133: if (toWrite != 0 && bufferList.size() != 0) {
134: byteBuffer = (ByteBuffer) bufferList.remove(bufferList
135: .size() - 1);
136: if (byteBuffer.remaining() == 0) {
137: bufferList.add(byteBuffer);
138: byteBuffer = null;
139: }
140: }
141:
142: while (toWrite != 0) {
143: try {
144: if (byteBuffer == null) {
145: byteBuffer = (ByteBuffer) handler.getServer()
146: .getByteBufferPool().borrowObject();
147: }
148: } catch (Exception e) {
149: logger.warning("Could not borrow ByteBufer from pool: "
150: + e);
151: throw new IOException(e.toString());
152: }
153:
154: remaining = byteBuffer.remaining();
155: if (remaining < toWrite) {
156: byteBuffer.put(b, off, remaining);
157: off = off + remaining;
158:
159: toWrite = toWrite - remaining;
160: } else {
161: byteBuffer.put(b, off, toWrite);
162: toWrite = 0;
163: }
164: bufferList.add(byteBuffer);
165: byteBuffer = null;
166: }
167: }
168:
169: //returns flag indicating if full write was done
170: public synchronized boolean writeAllByteBuffer() throws IOException {
171: if (lastByteBuffer != null) {
172: writeLastByteBuffer();
173: if (lastByteBuffer != null)
174: return false;
175: }
176:
177: if (bufferList.size() == 0) {
178: if (toNotify != null) {
179: synchronized (toNotify) {
180: toNotify.notify();
181: toNotify = null;
182: }
183: }
184: return true;
185: }
186:
187: while (bufferList.size() != 0) {
188: lastByteBuffer = (ByteBuffer) bufferList.remove(0);
189: lastByteBuffer.flip();
190: writeLastByteBuffer();
191: if (lastByteBuffer != null)
192: return false;
193: }
194:
195: if (toNotify != null) {
196: synchronized (toNotify) {
197: toNotify.notify();
198: toNotify = null;
199: }
200: }
201: return true;
202: }
203:
204: private synchronized void writeLastByteBuffer() throws IOException {
205: int written = 0;
206: while (lastByteBuffer.remaining() != 0) {
207: java.nio.channels.SocketChannel sc = handler
208: .getSocketChannel();
209: if (sc != null && sc.isOpen()) {
210: written = sc.write(lastByteBuffer);
211: if (written == 0) {
212: break;
213: }
214: if (logger.isLoggable(Level.FINEST)) {
215: logger.finest("Written " + written + " bytes");
216: }
217: } else {
218: throw new IOException("SocketChannel was closed.");
219: }
220: }
221: if (lastByteBuffer.remaining() == 0) {
222: returnBufferBack(lastByteBuffer);
223: lastByteBuffer = null;
224: }
225: }
226:
227: private void returnBufferBack(ByteBuffer byteBuffer) {
228: try {
229: handler.getServer().getByteBufferPool().returnObject(
230: byteBuffer);
231: } catch (Exception er) {
232: logger.warning("Error while returning ByteBuffer to pool: "
233: + er);
234: }
235: }
236:
237: public void forceNotify() {
238: if (toNotify == null)
239: return;
240: synchronized (toNotify) {
241: toNotify.notify();
242: toNotify = null;
243: }
244: }
245:
246: public boolean isDataAvailableForWrite(Object toNotify) {
247: if (lastByteBuffer != null) {
248: if (this .toNotify != null) {
249: throw new IllegalStateException(
250: "toNotify object was already set!");
251: }
252: this .toNotify = toNotify;
253: return true;
254: }
255: if (bufferList.size() == 0) {
256: return false;
257: } else {
258: if (this .toNotify != null) {
259: throw new IllegalStateException(
260: "toNotify object was already set!");
261: }
262: this .toNotify = toNotify;
263: return true;
264: }
265: }
266: }
|