java.io.Piped*Stream: read throws "Write end dead", but readable data remains

Name: diC59631 Date: 04/21/98


This occurs in 1.2beta4 build A.

PipedInputStream.read throws IOException saying WriteEndDead
if the writing thread is no longer alive, EVEN if there
are still unread bytes in the buffer written before the writing
thread terminated. This means that these last remaining bytes
cannot be read.

import java.io.*;

public class T421 {
  public static void main(String[] args) {
    PipedChannel p = new PipedChannel();
    Producer w = new Producer(p);
    Consumer c = new Consumer(p);
    new Thread(w).start();
    new Thread(c).start();
  }
}


class Record implements Serializable {
  String name = "My Name";
  String address = "My address";
  int age = 2;
}

class Producer implements Runnable {
  PipedChannel p;
  Producer(PipedChannel pp) { p = pp; }

  public void run() {
    Record r = new Record();
    try {
      for (int i = 0; i < 1000; ++i)
        p.put(r);
    }
    catch(Exception ex) {
      ex.printStackTrace();
      return;
    }
  }
}

class Consumer implements Runnable {
  PipedChannel p;
  Consumer(PipedChannel pp) { p = pp; }

  public void run() {
    try {
      for (int i = 0; i < 1000; ++i) {
        Record r = (Record)(p.take());
        Thread.sleep(2);
      }
    }
    catch(Exception ex) {
      ex.printStackTrace();
      return;
    }
  }
}




class PipedChannel {
  protected ObjectInputStream in_;
  protected ObjectOutputStream out_;
  protected PipedOutputStream outp_;
  protected PipedInputStream inp_;
  
  public PipedChannel() {
    try {
      outp_ = new PipedOutputStream();
      inp_ = new PipedInputStream();
      inp_.connect(outp_);
    }
    catch (IOException ex) {
      ex.printStackTrace();
      throw new Error("Cannot construct Pipe?");
    }
  }

  // needed since constructor can block on reads

  protected synchronized ObjectInputStream in() {
    try {
      if (in_ == null) in_ = new ObjectInputStream(inp_);
      return in_;
    }
    catch (IOException ex) { ex.printStackTrace(); return null; }
  }

  protected synchronized ObjectOutputStream out() {
    try {
      if (out_ == null) out_ = new ObjectOutputStream(outp_);
      return out_;
    }
    catch (IOException ex) { ex.printStackTrace(); return null; }
  }


  protected void doPut(Object x) throws InterruptedException {
    try {
      out().writeObject(x);
    }
    catch (InterruptedIOException ex) {
      ex.printStackTrace();
      throw new InterruptedException();
    }
    catch (IOException ex) {
      ex.printStackTrace();
      throw new InterruptedException();
    }
  }

 protected Object doTake() throws InterruptedException {
    try {
      return in().readObject();
    }
    catch (InterruptedIOException ex) {
      ex.printStackTrace();
      throw new InterruptedException();
    }
    catch (IOException ex) {
      ex.printStackTrace();
      throw new InterruptedException();
    }
    catch (ClassNotFoundException ex) {
      ex.printStackTrace();
      throw new InterruptedException();
    }
 }

  public void put(Object x) throws InterruptedException {
    if (x == null) throw new IllegalArgumentException();
    doPut(x);
  }

 public Object take() throws InterruptedException {
   Object x = null;
   x = doTake();
   return x;
 }



}
(Review ID: 28607)
======================================================================