GUACAMOLE-44: Allow intercepted streams to report errors.

guacamole/src/main/java/org/apache/guacamole/rest/tunnel/TunnelRESTService.java

@@ -136,7 +136,12 @@ public class TunnelRESTService {
 
             @Override
             public void write(OutputStream output) throws IOException {
-                tunnel.interceptStream(streamIndex, output);
+                try {
+                    tunnel.interceptStream(streamIndex, output);
+                }
+                catch (GuacamoleException e) {
+                    throw new IOException(e);
+                }
             }
 
         };
@@ -167,8 +172,9 @@ public class TunnelRESTService {
      *     stream.
      *
      * @throws GuacamoleException
-     *     If the session associated with the given auth token cannot be
+     *     If the session associated with the given auth
-     *     retrieved, or if no such tunnel exists.
+     *     token cannot be retrieved, if no such tunnel exists, or if the
+     *     intercepted stream itself closes with an error.
      */
     @POST
     @Consumes(MediaType.WILDCARD)

guacamole/src/main/java/org/apache/guacamole/tunnel/InputStreamInterceptingFilter.java

@@ -27,6 +27,7 @@ import javax.xml.bind.DatatypeConverter;
 import org.apache.guacamole.GuacamoleException;
 import org.apache.guacamole.net.GuacamoleTunnel;
 import org.apache.guacamole.protocol.GuacamoleInstruction;
+import org.apache.guacamole.protocol.GuacamoleStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -164,8 +165,24 @@ public class InputStreamInterceptingFilter
 
         // Terminate stream if an error is encountered
         if (!status.equals("0")) {
+
+            // Parse status code as integer
+            int code;
+            try {
+                code = Integer.parseInt(status);
+            }
+
+            // Assume internal error if parsing fails
+            catch (NumberFormatException e) {
+                logger.debug("Translating invalid status code \"{}\" to SERVER_ERROR.", status);
+                code = GuacamoleStatus.SERVER_ERROR.getGuacamoleStatusCode();
+            }
+
+            // Flag error and close stream
+            stream.setStreamError(code, args.get(1));
             closeInterceptedStream(stream);
             return;
+
         }
 
         // Send next blob

guacamole/src/main/java/org/apache/guacamole/tunnel/InterceptedStream.java

@@ -20,6 +20,8 @@
 package org.apache.guacamole.tunnel;
 
 import java.io.Closeable;
+import org.apache.guacamole.GuacamoleException;
+import org.apache.guacamole.protocol.GuacamoleStatus;
 
 /**
  * A simple pairing of the index of an intercepted Guacamole stream with the
@@ -45,6 +47,13 @@ public class InterceptedStream<T extends Closeable> {
      */
     private final T stream;
 
+    /**
+     * The exception which prevented the stream from completing successfully,
+     * if any. If the stream completed successfully, or has not encountered any
+     * exception yet, this will be null.
+     */
+    private GuacamoleException streamError = null;
+
     /**
      * Creates a new InterceptedStream which associated the given Guacamole
      * stream index with the given stream object.
@@ -83,4 +92,70 @@ public class InterceptedStream<T extends Closeable> {
         return stream;
     }
 
+    /**
+     * Reports that this InterceptedStream did not complete successfully due to
+     * the given GuacamoleException, which could not be thrown at the time due
+     * to asynchronous handling of the stream contents.
+     *
+     * @param streamError
+     *     The exception which prevented the stream from completing
+     *     successfully.
+     */
+    public void setStreamError(GuacamoleException streamError) {
+        this.streamError = streamError;
+    }
+
+    /**
+     * Reports that this InterceptedStream did not complete successfully due to
+     * an error described by the given status code and human-readable message.
+     * The error reported by this call can later be retrieved as a
+     * GuacamoleStreamException by calling getStreamError().
+     *
+     * @param code
+     *     The Guacamole protocol status code which described the error that
+     *     occurred. This should be taken directly from the "ack" instruction
+     *     that reported the error witin the intercepted stream.
+     *
+     * @param message
+     *     A human-readable message describing the error that occurred. This
+     *     should be taken directly from the "ack" instruction that reported
+     *     the error witin the intercepted stream.
+     */
+    public void setStreamError(int code, String message) {
+
+        // Map status code to GuacamoleStatus, assuming SERVER_ERROR by default
+        GuacamoleStatus status = GuacamoleStatus.fromGuacamoleStatusCode(code);
+        if (status == null)
+            status = GuacamoleStatus.SERVER_ERROR;
+
+        // Associate stream with corresponding GuacamoleStreamException
+        setStreamError(new GuacamoleStreamException(status, message));
+
+    }
+
+    /**
+     * Returns whether an error has prevented this InterceptedStream from
+     * completing successfully. This will return false if the stream has
+     * completed successfully OR if the stream simply has not yet completed.
+     *
+     * @return
+     *     true if an error has prevented this InterceptedStream from
+     *     completing successfully, false otherwise.
+     */
+    public boolean hasStreamError() {
+        return streamError != null;
+    }
+
+    /**
+     * Returns a GuacamoleException which describes why this InterceptedStream
+     * did not complete successfully.
+     *
+     * @return
+     *     An exception describing the error that prevented the stream from
+     *     completing successfully, or null if no such error has occurred.
+     */
+    public GuacamoleException getStreamError() {
+        return streamError;
+    }
+
 }

guacamole/src/main/java/org/apache/guacamole/tunnel/StreamInterceptingFilter.java

@@ -177,8 +177,12 @@ public abstract class StreamInterceptingFilter<T extends Closeable>
      * @param stream
      *     The stream object which will produce or consume all data for the
      *     stream having the given index.
+     *
+     * @throws GuacamoleException
+     *     If an error occurs while intercepting the stream, or if the stream
+     *     itself reports an error.
      */
-    public void interceptStream(int index, T stream) {
+    public void interceptStream(int index, T stream) throws GuacamoleException {
 
         InterceptedStream<T> interceptedStream;
         String indexString = Integer.toString(index);
@@ -204,6 +208,10 @@ public abstract class StreamInterceptingFilter<T extends Closeable>
         // Wait for stream to close
         streams.waitFor(interceptedStream);
 
+        // Throw any asynchronously-provided exception
+        if (interceptedStream.hasStreamError())
+            throw interceptedStream.getStreamError();
+
     }
 
 }

guacamole/src/main/java/org/apache/guacamole/tunnel/StreamInterceptingTunnel.java

@@ -85,18 +85,27 @@ public class StreamInterceptingTunnel extends DelegatingGuacamoleTunnel {
      *
      * @param stream
      *     The OutputStream to write all intercepted data to.
+     *
+     * @throws GuacamoleException
+     *     If an error occurs while intercepting the stream, or if the stream
+     *     itself reports an error.
      */
-    public void interceptStream(int index, OutputStream stream) {
+    public void interceptStream(int index, OutputStream stream)
+            throws GuacamoleException {
 
         // Log beginning of intercepted stream
         logger.debug("Intercepting output stream #{} of tunnel \"{}\".",
                 index, getUUID());
 
-        outputStreamFilter.interceptStream(index, new BufferedOutputStream(stream));
+        try {
+            outputStreamFilter.interceptStream(index, new BufferedOutputStream(stream));
+        }
 
         // Log end of intercepted stream
-        logger.debug("Intercepted output stream #{} of tunnel \"{}\" ended.",
+        finally {
-                index, getUUID());
+            logger.debug("Intercepted output stream #{} of tunnel \"{}\" ended.",
+                    index, getUUID());
+        }
 
     }
 
@@ -113,18 +122,27 @@ public class StreamInterceptingTunnel extends DelegatingGuacamoleTunnel {
      *
      * @param stream
      *     The InputStream to read all blobs data from.
+     *
+     * @throws GuacamoleException
+     *     If an error occurs while intercepting the stream, or if the stream
+     *     itself reports an error.
      */
-    public void interceptStream(int index, InputStream stream) {
+    public void interceptStream(int index, InputStream stream)
+            throws GuacamoleException {
 
         // Log beginning of intercepted stream
         logger.debug("Intercepting input stream #{} of tunnel \"{}\".",
                 index, getUUID());
 
-        inputStreamFilter.interceptStream(index, new BufferedInputStream(stream));
+        try {
+            inputStreamFilter.interceptStream(index, new BufferedInputStream(stream));
+        }
 
         // Log end of intercepted stream
-        logger.debug("Intercepted input stream #{} of tunnel \"{}\" ended.",
+        finally {
-                index, getUUID());
+            logger.debug("Intercepted input stream #{} of tunnel \"{}\" ended.",
+                    index, getUUID());
+        }
 
     }