OutputStreamEventQueue


1   
2   /*
3    * Enhydra Java Application Server Project
4    * 
5    * The contents of this file are subject to the Enhydra Public License
6    * Version 1.1 (the "License"); you may not use this file except in
7    * compliance with the License. You may obtain a copy of the License on
8    * the Enhydra web site ( http://www.enhydra.org/ ).
9    * 
10   * Software distributed under the License is distributed on an "AS IS"
11   * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See 
12   * the License for the specific terms governing rights and limitations
13   * under the License.
14   * 
15   * The Initial Developer of the Enhydra Application Server is Lutris
16   * Technologies, Inc. The Enhydra Application Server and portions created
17   * by Lutris Technologies, Inc. are Copyright Lutris Technologies, Inc.
18   * All Rights Reserved.
19   * 
20   * Contributor(s):
21   * 
22   * $Id: OutputStreamEventQueue.java,v 1.2 2005/03/24 10:51:25 slobodan Exp $
23   */
24  
25  
26  
27  
28   
29  package com.lutris.util;
30  
31  import java.io.IOException  ;
32  import java.io.OutputStream  ;
33  import java.util.Date  ;
34  import java.util.Vector  ;
35  
36  
37  /**
38      A queue of "events". Each event is a write to the OutputStream.
39      This class implements both the reader and writer half of the queue. <P>
40  
41      Each "event" is returned as an OutputStreamEventQueueEntry object. <P>
42  
43      The "writer" interface is the set of OutputStream calls. This class
44      extends OutputStream, overriding all the OutputStream methods.
45      All data written to this object is stored up as "events" in an internal
46      queue. Each write is a separate event, so sometimes you will get one
47      event with data followed by one with just a newline. You can pass objects
48      of this class to anything that expects and OutputStream. Calls to the
49      OutputStream methods always return immediatly. <P>
50  
51      You may optionally specify a maximum number of bytes to store. After
52      each write to the queue, if the new total number of bytes is larger than
53      this limit, then the oldest entries are discarded until the total is
54      at or below the limit. This could result in throwing out all the elements
55      in the queue. Only the number of bytes of data written by the
56      <CODE>write()</CODE> methods is counted, not the additional date 
57      stamping and object encapsulation. If you specify a size limit of
58      zero, then no limit is imposed. <P>
59  
60      The "reader" interface is the getEvent() method. Calls to this method
61      will block until there is an event to return. Readers may also call the
62      hasEventsPending() method to see if there are any events. This call does
63      not block. <P>
64  
65      One possible use for this class is with the PrintTransactionFilter class.
66      If you pass in an instance of this class as the OutputStream, you may
67      then fetch the logging data in a convienent way. <P>
68  
69      @see com.lutris.util.OutputStreamEventQueueEntry
70      @see java.io.OutputStream
71      @see com.lutris.servlet.filter.PrintTransactionFilter
72      @author Andy John
73  */
74  public class OutputStreamEventQueue extends OutputStream   {
75  
76      // Is this "stream" open or closed? Don't allow writes to a closed stream.
77      private boolean     open;
78  
79      // My list of events.
80      private Vector        queue;
81  
82      // The current amount of data stored in the queue.
83      private int         numBytes;
84  
85      // The maximum number of bytes to be stores. Zero indicates no limit.
86      private int         maxBytes;
87  
88      /**
89          Create a new, empty, OutputStreamEventQueue with no limit on the
90          size of the data stored.
91      */
92      public OutputStreamEventQueue() {
93          this(0);
94      }
95      
96      /**
97          Create a new, empty, OutputStreamEventQueue with a limit on the
98          number of bytes it will store. 
99      */
100     public OutputStreamEventQueue(int maxBytes) {
101         this.maxBytes = maxBytes;
102         this.numBytes = 0;
103         open = true;
104         queue = new Vector  ();
105     }
106 
107     //--------------------------------------------------------------------
108     //
109     //          Writer functions
110     //
111     //--------------------------------------------------------------------
112 
113 
114     /**
115         Add an event to the queue containing one byte.
116         
117         @param b The byte to save in the queue.
118         @exception IOException If close() has been called.
119         @see java.io.OutputSream
120     */
121     public void write(int b) throws IOException   {
122         synchronized (queue) {
123             if (!open)
124                 throw new IOException  ("Wrote to closed OutputStreamEventQueue.");
125             OutputStreamEventQueueEntry evt = new OutputStreamEventQueueEntry();
126             evt.when = new Date  ();
127             evt.data = new byte[1];
128             evt.data[0] = (byte) b;
129             queue.addElement(evt);
130             this.numBytes += 4;
131             limitQueueSize();
132             queue.notify(); // Wake up one waiting reader.
133         }
134     }
135 
136 
137     /**
138         Add an event to the queue containing an array of bytes.
139         
140         @param b The array of bytes to save in the queue.
141         @exception IOException If close() has been called.
142         @see java.io.OutputSream
143     */
144     public void write(byte b[]) throws IOException   {
145         synchronized (queue) {
146             if (!open)
147                 throw new IOException  ("Wrote to closed OutputStreamEventQueue.");
148             OutputStreamEventQueueEntry evt = new OutputStreamEventQueueEntry(new Date  (), b);
149             queue.addElement(evt);
150             this.numBytes += b.length;
151             limitQueueSize();
152             queue.notify(); // Wake up one waiting reader.
153         }
154     }
155 
156 
157     /**
158         Add an event to the queue containing part of an array of bytes.
159         Bytes b[off]..b[off+len-1] are used.
160         
161         @param b The array of bytes to save part of in the queue.
162         @exception IOException If close() has been called.
163         @see java.io.OutputSream
164     */
165     public void write(byte b[],
166                    int off,
167                    int len) throws IOException   {
168         synchronized (queue) {
169             if (!open)
170                 throw new IOException  ("Wrote to closed OutputStreamEventQueue.");
171             OutputStreamEventQueueEntry evt = new OutputStreamEventQueueEntry();
172             evt.when = new Date  ();
173             byte[] c = new byte[len];
174             for (int i=0; i<len; i++)
175                 c[i] = b[i + off];
176             evt.data = c;
177             queue.addElement(evt);
178             this.numBytes += b.length;
179             limitQueueSize();
180             queue.notify(); // Wake up one waiting reader.
181         }
182     }
183 
184 
185     /**
186         Does nothing. There is no buffering; every write immediatly adds an
187         event to the queue. This method is provided because OutputStream
188         does.
189 
190         @exception IOException If close() has been called.
191         @see java.io.OutputSream
192     */
193     public void flush() throws IOException   {
194         synchronized (queue) {
195             if (!open)
196                 throw new IOException  ("Flushed a closed OutputStreamEventQueue.");
197         }
198     }
199 
200 
201     /**
202         Closes the OutputStream. It is no longer available for writing.
203         Any further writes will throw an IOException. <P>
204 
205         Closing the stream wakes up all the threads blocked on a read.
206         If the queue is empty, they will all notice this and return null.
207         This is your signal that there are no events and there never will be.
208 
209         @exception IOException If close() has already been called.
210         @see java.io.OutputSream
211     */
212     public void close() throws IOException   {
213         synchronized (queue) {
214             if (!open)
215                 throw new IOException  ("Closed an already closed OutputStreamEventQueue.");
216             open = false;
217             queue.notifyAll();  // Wake up all waiting readers, so
218                                 // the all notice we are closed.
219         }
220     }
221 
222 
223 
224     /*
225         Only call this if you are already synchronized on queue.
226         Throw out the oldest events untill we get down below our
227         limit on the amount of data we can hold.
228         This might leave the queue empty if the limit is small and
229         the newest entry has alot of data!
230     */
231     private void limitQueueSize() {
232         if (maxBytes == 0) 
233             return;
234         while (numBytes > maxBytes) {
235             if (queue.size() <= 0) {
236                 numBytes = 0;
237                 return;
238             }
239             OutputStreamEventQueueEntry evt =
240                 (OutputStreamEventQueueEntry) queue.firstElement();
241             queue.removeElement(evt);
242             numBytes -= evt.data.length;
243             if (numBytes < 0 )
244                 numBytes = 0;
245         }   
246     }
247 
248     
249     //--------------------------------------------------------------------
250     //
251     //          Reader functions
252     //
253     //--------------------------------------------------------------------
254 
255     /**
256         Get an event from the queue. This will block until there is an
257         event to return. The data about the event is stored in a
258         OutputStreamEventQueueEntry object. <P>
259 
260         If null is returned, that means there are no more events, and the
261         stream is closed, so no further events will be generated.
262 
263         @return A OutputStreamEventQueueEntry object describing the event,
264         or null if the queue is empty and no more events are possible.
265         @see com.lutris.util.OutputStreamEventQueueEntry
266     */
267     public OutputStreamEventQueueEntry getEvent() {
268         while (true) {
269             // Try to return an entry.
270             synchronized (queue) {
271                 // Are there currently any waiting entries?
272                 if (queue.size() > 0) {
273                     // Pop off the queue.
274                     OutputStreamEventQueueEntry evt = 
275                         (OutputStreamEventQueueEntry) queue.firstElement();
276                     queue.removeElement(evt);
277                     numBytes -= evt.data.length;
278                     if (numBytes < 0)
279                         numBytes = 0;
280                     return evt; 
281                 } else {
282                     // Is it possible that more entries could show up?
283                     if (!open) 
284                         return null;
285                 }
286                 // Since we didn't return something, wait.
287                 try {
288                     queue.wait();
289                 } catch (InterruptedException   e) {
290                 }
291             }
292         }
293     }
294 
295 
296     /**
297         Is there an event waiting in the queue right now? This call will
298         not block, however it is possible that another thread will take the
299         event before you do, so your call to getEvent may still block.
300     
301         @return True is there is an event waiting in the queue.
302     */
303     public boolean hasEventsPending() {
304         synchronized (queue) {
305             return (queue.size() > 0);
306         }
307     }
308 }
309 
310 
311
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags