Blockierender PipedInputStream

[Krabat]

Kurzversion meines Problems:
Ich lese aus einem PipedInputStream, der sobald er "leer" ist, nicht "-1" zurück gibt, sondern blockt und scheinbar auf neue Daten wartet. Wie kommt das / was kann man dagegen tun?

public void addExternalMessage(Level logLevel, PrintStream outputStream) {
    if (null == outputStream || null == logLevel) {
      throw new NullPointerException();
    }
    if (outputStream.checkError()) {
      this.getExternalLogger().warn("An error was raised in the PrintStream");
    }
    outputStream.flush();
    InputStreamReader in = new InputStreamReader(internalInputStream);
    int bufferSize = 1024;
    int charsRead = 0;
    char[] charBuffer = new char[bufferSize];
    StringBuffer stringBuffer = new StringBuffer();
    try {
      while ((charsRead = in.read(charBuffer, 0, bufferSize)) != -1) {
        this.getExternalLogger().debug("PipedInputStream: read "+charsRead+" chars.");
        if (charsRead == bufferSize) {
          stringBuffer.append(charBuffer);
        } else if (charsRead > 0 && charsRead < bufferSize){
          stringBuffer.append(new String(charBuffer).substring(0, charsRead));
          in.reset();
          break;//workaround, since the empty PipedInputStream blocks...
        } else {
          this.getExternalLogger().debug("PipedInputStream: reached undefined state!");
        }
      }
    } catch (IOException e) {
      this.getExternalLogger().error("I/O Error while reading from internalInputStream");
    }
    String message = stringBuffer.toString().trim();
    if (Utils.isEmptyString(message)) {
      this.getExternalLogger().warn("read from empty internalInputStream");
      return;
    }
    this.getExternalLogger().log(logLevel, message);
}

Weitere Hintergrundinformationen
Aus Kompabilitätsgrunden ermögliche ich über eine getPrintStream() Methode den Zugriff auf einen PrintStream (OutputStream), in den aufrufende Instanzen schreiben können. Da ich jedesmal eine Log Nachricht generieren möchte, wenn in den Stream geschrieben wurde, muss die aufrufende Instanz die Methode addExternalMessage(...) aufrufen.

Damit ich aus dem OutputStream lesen kann, habe ich den PrintStream mit PipedIn/OutputStreams verbunden. Es handelt es sich sowohl bei dem PrintStream (nicht unbedingt nötig) als auch bei dem PipedInputStream um Klassenattribute.

PipedOutputStream pipedOutputStream = new PipedOutputStream();
internalInputStream = new PipedInputStream(pipedOutputStream);	//Klassen Attribut
externalPrintStream = new PrintStream(pipedOutputStream);	//Klassen Attribut

 

[FArt]

Du kannst nichts dagegen tun, denn das ist das in der API Dokumentation beschriebene verhalten:

Reads the next byte of data from this piped input stream. The value byte is returned as an int in the range 0 to 255. If no byte is available because the end of the stream has been reached, the value -1 is returned. This method blocks until input data is available, the end of the stream is detected, or an exception is thrown. If a thread was providing data bytes to the connected piped output stream, but the thread is no longer alive, then an IOException is thrown.

Ein Stream ist nie "leer". Daten können ankommen oder ausbleiben. Wenn sie ausbleiben, blockiert die read-Methode. Wenn der Sender fertig ist, schließt er den Stream und der Lesende erhält als letzen Wert (wenn alle anderen Werte ausgelesen wurden) -1 zurück.

Annahme: du hast auch das folgende in der API Doku gelesen, verstanden und eingehalten ;-)

A piped input stream should be connected to a piped output stream; the piped input stream then provides whatever data bytes are written to the piped output stream. Typically, data is read from a PipedInputStream object by one thread and data is written to the corresponding PipedOutputStream by some other thread. Attempting to use both objects from a single thread is not recommended, as it may deadlock the thread. The piped input stream contains a buffer, decoupling read operations from write operations, within limits.

[EDIT]
Wolltest du evtl. zeilenweise lesen/schreiben? Dann wrappe deine Streams mit BufferedStreams und Readern und lese bzw. schreibe dann zweilenweise.
Was möchtest du mit flush() und reset() erreichen?

 

[Krabat]

Wolltest du evtl. zeilenweise lesen/schreiben?

Nun, eigentlich wollte ich alles auslesen, was im InputStream vorhanden ist.

Was möchtest du mit flush() und reset() erreichen?

Mit Flush wollte ich erreichen, dass eventuelle Puffer geleert werden und alles im InputStream "ankommt". Den "reset" hatte ich noch eingebaut, da bei erneutem Aufruf wieder die Blockierung einsetzte und ich dachte, der interne Zeiger müsse wieder auf den Beginn des Streams gesetzt werden, aber bei näherem Überlegen wird der Stream durch die read Methode bereits geleert...

Im übrigen greife ich von einem Thread aus schreibend (PrintStream) und anschließend lesend (PipedInputStream) auf die Streams zu. Wie kann ich geschickter aus einem OutputStream lesen, dessen Referenz an eine andere Instanz gegangen ist? Wie kann ich dem block entgegen wirken?

Auch verstehe ich nicht so ganz, wieso nicht "-1" zurückgegen wird, ich bin doch am Ende des Streams angelangt...

Ein Stream ist nie "leer". Daten können ankommen oder ausbleiben. Wenn sie ausbleiben, blockiert die read-Methode. Wenn der Sender fertig ist, schließt er den Stream und der Lesende erhält als letzen Wert (wenn alle anderen Werte ausgelesen wurden) -1 zurück.

Muss ich also den OutputStream schließen, statt ihn zu "flushen"?

 

[FArt]

Du hast noch ein Verständnisproblem mit Streams im Allgemeinen und mit den PipedStream im Besonderen.

Nun, eigentlich wollte ich alles auslesen, was im InputStream vorhanden ist.

Das hast du, sonst würde der Stream (wenn richtig angewendet) nicht blockieren.

Im übrigen greife ich von einem Thread aus schreibend (PrintStream) und anschließend lesend (PipedInputStream) auf die Streams zu.

Das zählt nicht zu "richtig angewendet". Bei PipedStreams müssen zwei Threads gleichzeitig beteiligt sein, nicht einer der die Sachen nacheinander ausführt. (siehe API Doku!!!).

Auch verstehe ich nicht so ganz, wieso nicht "-1" zurückgegen wird, ich bin doch am Ende des Streams angelangt...

Das Ende ist erreicht, wenn der Eingabestream geschlossen wurde, nicht wenn du glaubst fertig zu sein und keine Daten mehr reinschiebst.

Du kannst auch nach langen Pausen wieder Daten in einen offenen Stream reinschieben. Wenn du fertig bist, musst du den Stream explizit schließen.

Du kannst auch regelmäßig immer wieder Daten in einen Stream schieben, ohne diesen zu schließen. Dann muss aber für beide Seiten ein Begrenzungszeichen bekannt sein für eine Teillieferung der Daten... z.B. ein CR/LF... dann kann man z.B. zeilenweise lesen und schreiben. Zwischendurch blockiert der lesende und wartet auf die nächste Eingabe oder auf das Ende.

Gib dir mal speziell dies, besser noch das ganze Kapitel über Streams: http://www.galileodesign.de/openbook/javainsel5/javainsel12_010.htm#Rxx747java12010040004011F027100

 

[Krabat]

Ich habe nun eingesehen, dass der Fehler weniger in der Anwendung von PipedIn/OutputStream lag, als vielmehr diese überhaupt zu verwenden. Habe das ganze noch einmal umgebaut und verwende nun intern einen ByteArrayIn/OutputStream, womit es auch keine weiteren Fehler gibt. Bei Interesse poste ich gerne den neuen Quellcode.

FArt, vielen Dank für die guten Anmerkungen!