Rethrow StreamExceptions in Client

Instead of wrapping them in StreamExceptions. Easier to
deal with timeout exception.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Environment.java

@@ -42,7 +42,7 @@ static EnvironmentBuilder builder() {
               .getConstructor()
               .newInstance();
     } catch (Exception e) {
-      throw new StreamException(e);
+      throw new StreamException("Error while creating stream environment builder", e);
     }
   }
 

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -46,6 +46,7 @@
 import static com.rabbitmq.stream.impl.Utils.extractResponseCode;
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
 import static java.lang.String.format;
+import static java.lang.String.join;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import com.rabbitmq.stream.AuthenticationFailureException;
@@ -421,10 +422,11 @@ private Map<String, String> peerProperties() {
         throw new StreamException("Error when establishing stream connection", request.error());
       }
     } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while trying to exchange peer properties", e);
     }
   }
 
@@ -484,9 +486,12 @@ private SaslAuthenticateResponse sendSaslAuthenticate(
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while trying to authenticate", e);
     }
   }
 
@@ -516,7 +521,7 @@ private Map<String, String> open(String virtualHost) {
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error during open command", e);
     }
   }
 
@@ -527,7 +532,7 @@ void send(byte[] content) {
     try {
       channel.writeAndFlush(bb).sync();
     } catch (InterruptedException e) {
-      throw new StreamException(e);
+      throw new StreamException("Error while sending bytes", e);
     }
   }
 
@@ -555,9 +560,12 @@ private void sendClose(short code, String reason) {
             "Unexpected response code when closing: "
                 + formatConstant(request.response.get().getResponseCode()));
       }
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while closing connection", e);
     }
   }
 
@@ -575,9 +583,12 @@ private List<String> getSaslMechanisms() {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while exchanging SASL mechanisms", e);
     }
   }
 
@@ -611,9 +622,12 @@ public Response create(String stream, Map<String, String> arguments) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(format("Error while creating stream '%s'", stream), e);
     }
   }
 
@@ -656,9 +670,12 @@ public Response delete(String stream) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(format("Error while deleting stream '%s'", stream), e);
     }
   }
 
@@ -688,9 +705,13 @@ public Map<String, StreamMetadata> metadata(String... streams) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format("Error while getting metadata for stream(s) '%s'", join(",", streams)), e);
     }
   }
 
@@ -728,7 +749,8 @@ public Response declarePublisher(byte publisherId, String publisherReference, St
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format("Error while declaring publisher for stream '%s'", stream), e);
     }
   }
 
@@ -752,7 +774,7 @@ public Response deletePublisher(byte publisherId) {
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while deleting publisher", e);
     }
   }
 
@@ -1090,7 +1112,8 @@ public Response subscribe(
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format("Error while trying to subscribe to stream '%s'", stream), e);
     }
   }
 
@@ -1147,7 +1170,10 @@ public QueryOffsetResponse queryOffset(String reference, String stream) {
       throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format(
+              "Error while querying offset for reference '%s' on stream '%s'", reference, stream),
+          e);
     }
   }
 
@@ -1185,9 +1211,16 @@ public long queryPublisherSequence(String publisherReference, String stream) {
             formatConstant(response.getResponseCode()));
       }
       return response.getSequence();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format(
+              "Error while querying publisher sequence for '%s' on stream '%s'",
+              publisherReference, stream),
+          e);
     }
   }
 
@@ -1206,9 +1239,12 @@ public Response unsubscribe(byte subscriptionId) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while unsubscribing", e);
     }
   }
 
@@ -1330,9 +1366,16 @@ public List<String> route(String routingKey, String superStream) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format(
+              "Error while querying route for routing key '%s' on super stream '%s'",
+              routingKey, superStream),
+          e);
     }
   }
 
@@ -1356,9 +1399,13 @@ public List<String> partitions(String superStream) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format("Error while querying partitions for super stream '%s'", superStream), e);
     }
   }
 
@@ -1384,9 +1431,12 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException("Error while exchanging command version", e);
     }
   }
 
@@ -1409,9 +1459,13 @@ StreamStatsResponse streamStats(String stream) {
       channel.writeAndFlush(bb);
       request.block();
       return request.response.get();
+    } catch (StreamException e) {
+      outstandingRequests.remove(correlationId);
+      throw e;
     } catch (RuntimeException e) {
       outstandingRequests.remove(correlationId);
-      throw new StreamException(e);
+      throw new StreamException(
+          format("Error while querying statistics for stream '%s'", stream), e);
     }
   }
 

src/main/java/com/rabbitmq/stream/impl/Codecs.java

@@ -31,7 +31,7 @@ private static Codec instanciateDefault() {
       return (Codec)
           Class.forName("com.rabbitmq.stream.codec.QpidProtonCodec").getConstructor().newInstance();
     } catch (Exception e) {
-      throw new StreamException(e);
+      throw new StreamException("Error while creating QPid Proton codec", e);
     }
   }
 }

src/main/java/com/rabbitmq/stream/impl/CompressionCodecs.java

@@ -33,7 +33,7 @@ private static CompressionCodecFactory instanciateDefault() {
               .getConstructor()
               .newInstance();
     } catch (Exception e) {
-      throw new StreamException(e);
+      throw new StreamException("Error while creating compression codec factory", e);
     }
   }
 }