experimental thread-killing capabilities in Stream
[org.ibex.io.git] / src / org / ibex / io / Stream.java
1 // Copyright 2004 Adam Megacz, see the COPYING file for licensing [LGPL]
2 package org.ibex.io;
3
4 import java.io.*;
5 import java.net.*;
6 import java.util.*;
7 import java.util.zip.*;
8 import org.ibex.util.*;
9
10 /** plays the role of InputStream, OutputStream, Reader and Writer, with logging and unchecked exceptions */
11 public class Stream {
12
13     protected final In in;
14     protected final Out out;
15     private         StringBuffer log = loggingEnabled ? new StringBuffer(16 * 1024) : null;
16     private         String newLine = "\r\n";
17
18     public static boolean loggingEnabled = System.getProperty("ibex.io.stream.logEnabled", "true") != null;
19
20     public  Stream(InputStream in)                   { this.in = new Stream.In(in); this.out = null; }
21     public  Stream(                OutputStream out) { this.in = null;              this.out = new Stream.Out(out); }
22     public  Stream(InputStream in, OutputStream out) { this.in = new Stream.In(in); this.out = new Stream.Out(out); }
23     public  Stream(String s)                         { this(new ByteArrayInputStream(s.getBytes())); }
24     public  Stream(Socket s) {
25         try { this.in = new Stream.In(s.getInputStream());    } catch (IOException e) { throw new StreamException(e); }
26         try { this.out = new Stream.Out(s.getOutputStream()); } catch (IOException e) { throw new StreamException(e); }
27     }
28
29     private static int ioe(Exception e) { throw new StreamException(e); }
30     public static class StreamException extends RuntimeException {
31         public StreamException(Exception e) { super(e); }
32         public StreamException(String s)    { super(s); }
33     }
34     public static class EOF             extends StreamException  { public EOF() { super("End of stream"); } }
35     public static class Closed          extends StreamException  { public Closed(String s) { super(s); } }
36
37
38     private static Hashtable blocker = new Hashtable();
39     public static void kill(Thread thread) {
40         Stream block = (Stream)blocker.get(thread);
41         if (block == null) {
42             Log.warn(Stream.class, "thread " + thread + " is not blocked on a stream");
43         } else {
44             Log.warn(Stream.class, "asynchronously closing " + block);
45             block.close();
46         }
47     }
48
49     public char   peekc()                          {
50         Stream old = (Stream)blocker.get(Thread.currentThread());
51         try {
52             blocker.put(Thread.currentThread(), this);
53             flush(); return in.getc(true);
54         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
55     }
56     public char   getc()                           {
57         Stream old = (Stream)blocker.get(Thread.currentThread());
58         try {
59             blocker.put(Thread.currentThread(), this);
60             flush(); char ret = in.getc(false); log(ret); return ret;
61         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
62     }   
63     public String readln()                         {
64         Stream old = (Stream)blocker.get(Thread.currentThread());
65         try {
66             blocker.put(Thread.currentThread(), this);
67             flush(); String s = in.readln(); log(s); log('\n'); return s;
68         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
69     }
70     
71     public void   print(String s) {
72         Stream old = (Stream)blocker.get(Thread.currentThread());
73         try {
74             blocker.put(Thread.currentThread(), this);
75             logWrite(s);
76             out.write(s);
77             flush();
78         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
79     }
80     public void   println(String s) {
81         Stream old = (Stream)blocker.get(Thread.currentThread());
82         try {
83             blocker.put(Thread.currentThread(), this);
84             logWrite(s);
85             out.write(s);
86             out.write(newLine);
87             flush();
88         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
89     }
90     public void   flush()                          {
91         Stream old = (Stream)blocker.get(Thread.currentThread());
92         try {
93             blocker.put(Thread.currentThread(), this);
94             if (out != null) try { out.w.flush(); } catch(IOException e) { ioe(e); }
95         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
96     }
97     public int    read(byte[] b, int off, int len) {
98         Stream old = (Stream)blocker.get(Thread.currentThread());
99         try {
100             blocker.put(Thread.currentThread(), this);
101             flush();
102             int ret = in.readBytes(b, off, len);
103             if (log != null) log("\n[read " + ret + " bytes of binary data ]\n");
104             nnl = false;
105             return ret;
106         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
107     }
108     public int    read(char[] c, int off, int len) {
109         Stream old = (Stream)blocker.get(Thread.currentThread());
110         try {
111             blocker.put(Thread.currentThread(), this);
112             flush();
113             int ret = in.read(c, off, len);
114             if (log != null && ret != -1) log(new String(c, off, ret));
115             return ret;
116         } finally { if (old == null) blocker.remove(Thread.currentThread()); else blocker.put(Thread.currentThread(), old); }
117     }
118
119     public void   unread(String s)                 { in.unread(s); }
120
121     public void   close()                          { try { in.close(); } finally { out.close(); } }
122     public void   setNewline(String s)             { newLine = s; }
123
124
125     /** dumps the connection log into a file */
126     public  String dumpLog()         { String ret = log.toString(); log = new StringBuffer(16 * 1024); return ret; }
127     private void  log(String s)      { if(log==null) return; if (!nnl) Log.note("\n[read ] "); Log.note(s + "\n"); nnl=false; if (log != null) log.append(s); }
128     private void  logWrite(String s) { if(log==null) return; if (nnl) Log.note("\n"); Log.note("[write] "+s+"\n"); nnl=false; if (log != null) log.append(s); }
129     private void  log(char c)        { if(log==null) return; if (c == '\r') return; if (!nnl) Log.note("[read ] "); Log.note(c+""); nnl = c != '\n'; if (log != null) log.append(c); }
130     private boolean nnl = false;
131
132     private static class Out extends BufferedOutputStream {
133         private Writer w = new BufferedWriter(new OutputStreamWriter(this));
134         public Out(OutputStream out) { super(out); }
135         public void close() { try { super.close(); } catch (Exception e) { Log.error(this, e); } }
136         public void write(String s) { try { w.write(s); } catch (IOException e) { ioe(e); } }
137     }
138
139     private static class In extends InputStream {
140         public Reader reader = new InputStreamReader(this);
141         private final InputStream orig;
142         public In(InputStream in) { orig = in; }
143
144         char[] cbuf = new char[8192];
145         int cstart = 0;
146         int cend = 0;
147
148         byte[] buf = new byte[8192];
149         int start = 0;
150         int end = 0;
151
152         boolean flushing = false;
153
154         public int available() { return flushing ? 0 : (end - start); }
155         public void close() { try { orig.close(); } catch (Exception e) { Log.error(this, e); } }
156
157         public char getc(boolean peek) { try {
158             if (cstart == cend) {
159                 cstart = 0;
160                 cend = reader.read(cbuf, 0, cbuf.length);
161                 if (cend == -1) { cend = cstart; throw new EOF(); }
162             }
163             return peek ? cbuf[cstart] : cbuf[cstart++];
164         } catch (IOException e) { return (char)ioe(e); } }
165
166         public String readln() { try {
167             while(true) {
168                 for(int i=cstart; i<cend; i++)
169                     if (cbuf[i] == '\n') {
170                         // this should (in theory) handle CR, LF,
171                         // CRLF, and LFCR properly, assuming that the
172                         // file consistently uses the same ending
173                         // throughout.
174                         int begin = cstart;
175                         int len = i-cstart;
176                         cstart = i+1;
177                         if (cbuf[begin] == '\r') { begin++; len--; }
178                         while (len > 0 && cbuf[begin+len-1] == '\r') { len--; }
179                         return new String(cbuf, begin, len);
180                     }
181                 ensurec(256);
182                 int numread = reader.read(cbuf, cend, cbuf.length - cend);
183                 if (numread == -1) {
184                     if (cstart == cend) return null;
185                     String ret = new String(cbuf, cstart, cend-cstart);
186                     cstart = cend = 0;
187                     return ret;
188                 }
189                 cend += numread;
190             }
191         } catch (IOException e) { ioe(e); return null; } }
192
193         public int read(char[] c, int pos, int len) { try {
194             if (cstart == cend) {
195                 cstart = 0;
196                 cend = reader.read(cbuf, 0, cbuf.length);
197                 if (cend == -1) { cend = cstart; return -1; }
198             }
199             if (len > cend - cstart) len = cend - cstart;
200             System.arraycopy(cbuf, cstart, c, pos, len);
201             cstart += len;
202             return len;
203         } catch (IOException e) { ioe(e); return -1; } }
204
205         public int readBytes(byte[] b, int pos, int len) { flushchars(); return read(b, pos, len); }
206         public int read() { byte[] b = new byte[1]; if (read(b, 0, 1) == -1) return -1; return (int)b[0]; } 
207         public int read(byte[] b, int pos, int len) { try {
208             if (start == end) {
209                 start = 0;
210                 end = orig.read(buf, 0, buf.length);
211                 if (end == -1) { end = start; return -1; }
212             }
213             if (len > end - start) len = end - start;
214             System.arraycopy(buf, start, b, pos, len);
215             start += len;
216             return len;
217         } catch (IOException e) { ioe(e); return -1; } }
218
219         private void growc(int s){char[] cbuf2=new char[cbuf.length+s*2];System.arraycopy(cbuf,0,cbuf2,0,cbuf.length);cbuf=cbuf2; }
220         private void shiftc() {
221             char[] cbuf2 = new char[cbuf.length];
222             System.arraycopy(cbuf, cstart, cbuf2, 0, cend-cstart);
223             cend -= cstart;
224             cstart = 0;
225             cbuf = cbuf2;
226         }
227         private void ensurec(int space) { if (cend-cstart+space>cbuf.length) growc(space); if (cend+space>cbuf.length) shiftc(); }
228
229         private void growb(int s) { byte[] buf2 = new byte[buf.length+s*2]; System.arraycopy(buf,0,buf2,0,buf.length); buf=buf2; }
230         private void shiftb() { System.arraycopy(buf, start, buf, 0, end-start); end -= start; start = 0; }
231         private void ensureb(int space) { if (end-start+space>buf.length) growb(space); if (end+space>buf.length) shiftb(); }
232         private void ensureb2(int space) { if (end-start+space>buf.length) growb(space); if (start<space) unshiftb(); }
233         private void unshiftb() {
234             System.arraycopy(buf,start,buf,buf.length-(end-start),end-start);start=buf.length-(end-start);end=buf.length; }
235
236         public  void unread(String s) { ensurec(s.length()); s.getChars(0, s.length(), cbuf, cend); cend += s.length(); }
237
238         private void flushchars() {
239             try {
240                 flushing = true;
241                 for(; reader.ready(); reader.read(cbuf, cend++, 1)) ensurec(1);
242                 unreader.write(cbuf, cstart, cend);
243                 cstart = cend = 0;
244                 unreader.flush();
245             } catch (IOException e) { ioe(e);
246             } finally { flushing = false; }
247         }
248
249         Writer unreader = new OutputStreamWriter(new OutputStream() {
250                 public void close() { }
251                 public void write(int i) throws IOException { byte[] b = new byte[1]; b[0] = (byte)i; write(b, 0, 1); }
252                 public void write(byte[] b) throws IOException { write(b, 0, b.length); }
253                 public void write(byte[] b, int p, int l) {
254                     ensureb2(l);
255                     System.arraycopy(b, p, buf, start-l, l);
256                     start -= l;
257                 }
258             });
259     }
260 }