src/jrw/util/LazyPChannel.java

   1 package jrw.util;
   2
   3 import jrw.*;
   4 import java.nio.*;
   5 import java.nio.channels.*;
   6 import java.nio.charset.*;
   7
   8 public abstract class LazyPChannel implements ReadableByteChannel {
   9     private ByteBuffer curbuf = null;
  10     private boolean eof = false;
  11     private CharsetEncoder enc = null;
  12     private Runnable rem = null;
  13
  14     protected boolean write(byte[] data, int off, int len) {
  15         if(rem != null) throw(new IllegalStateException("buffer filled"));
  16         int t = Math.min(curbuf.remaining(), len);
  17         curbuf.put(data, off, t);
  18         if(len > t) {
  19             rem = () -> write(data, off + t, len - t);
  20             return(true);
  21         }
  22         return(false);
  23     }
  24     protected boolean write(byte[] data) {return(write(data, 0, data.length));}
  25
  26     protected boolean write(CharBuffer buf) {
  27         if(rem != null) throw(new IllegalStateException("buffer filled"));
  28         if(enc == null)
  29             enc = charset().newEncoder();
  30         while(true) {
  31             int pp = buf.position();
  32             CoderResult res = enc.encode(buf, curbuf, false);
  33             if(buf.remaining() == 0)
  34                 return(false);
  35             if(res.isUnderflow()) {
  36                 if(pp == buf.position()) {
  37                     /* XXX? Not sure if this can be expected to
  38                      * happen. I'm not aware of any charsets that should
  39                      * require it, and it would complicate the design
  40                      * significantly. */
  41                     throw(new RuntimeException("encoder not consuming input"));
  42                 }
  43             } else if(res.isOverflow()) {
  44                 rem = () -> write(buf);
  45                 return(true);
  46             } else {
  47                 try {
  48                     res.throwException();
  49                 } catch(CharacterCodingException e) {
  50                     throw(new RuntimeException(e));
  51                 }
  52             }
  53         }
  54     }
  55
  56     protected boolean write(CharSequence chars) {
  57         CharBuffer buf = (chars instanceof CharBuffer) ? ((CharBuffer)chars).duplicate() : CharBuffer.wrap(chars);
  58         return(write(buf));
  59     }
  60
  61     private void encflush2() {
  62         while(true) {
  63             CoderResult res = enc.flush(curbuf);
  64             if(res.isOverflow()) {
  65                 rem = this::encflush1;
  66                 return;
  67             } else if(res.isUnderflow()) {
  68                 return;
  69             } else {
  70                 try {
  71                     res.throwException();
  72                 } catch(CharacterCodingException e) {
  73                     throw(new RuntimeException(e));
  74                 }
  75             }
  76         }
  77     }
  78
  79     private void encflush1() {
  80         CharBuffer empty = CharBuffer.wrap("");
  81         while(true) {
  82             CoderResult res = enc.encode(empty, curbuf, true);
  83             if(res.isOverflow()) {
  84                 rem = this::encflush1;
  85                 return;
  86             } else if(res.isUnderflow()) {
  87                 rem = this::encflush2;
  88                 return;
  89             } else {
  90                 try {
  91                     res.throwException();
  92                 } catch(CharacterCodingException e) {
  93                     throw(new RuntimeException(e));
  94                 }
  95             }
  96         }
  97     }
  98
  99     private void encflush() {
 100         if(enc != null)
 101             rem = this::encflush1;
 102     }
 103
 104     protected Charset charset() {return(Http.UTF8);}
 105
 106     protected abstract boolean produce();
 107
 108     public int read(ByteBuffer buf) {
 109         curbuf = buf;
 110         try {
 111             int op = buf.position();
 112             while(buf.remaining() > 0) {
 113                 Runnable rem = this.rem;
 114                 this.rem = null;
 115                 if(rem != null) {
 116                     rem.run();
 117                 } else {
 118                     if(eof) {
 119                         break;
 120                     } else if(produce()) {
 121                         encflush();
 122                         eof = true;
 123                     }
 124                 }
 125             }
 126             if(eof && (buf.position() == op))
 127                 return(-1);
 128             return(buf.position() - op);
 129         } finally {
 130             curbuf = null;
 131         }
 132     }
 133
 134     public void close() {}
 135     public boolean isOpen() {return(true);}
 136 }