Propagate the exception thrown from the callback

Signed-off-by: Rafael Luis Ibasco <[email protected]>

Author: ribasco

examples/src/main/resources/logback.xml

@@ -91,5 +91,4 @@
 
     <logger name="io.netty" level="error"/>
     <logger name="com.ibasco" level="error"/>
-    <logger name="org.jline" level="debug"/>
 </configuration>

protocols/valve/steam/master/src/main/java/com/ibasco/agql/protocols/valve/steam/master/MasterServerMessenger.java

@@ -54,8 +54,10 @@
 import java.time.Duration;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -106,6 +108,9 @@ private void initFailSafe(final Options options, final ScheduledExecutorService
                 return new MasterServerResponse(new HashSet<>(timeoutException.getAddresses()));
             }
             return new MasterServerResponse(new HashSet<>());
+        }).handleIf(p -> {
+            Throwable cause = Errors.unwrap(p);
+            return !(cause instanceof CancellationException) && !(cause instanceof InterruptedException) && !(cause instanceof RejectedExecutionException);
         }).build();
 
         //retry policy
@@ -130,6 +135,10 @@ private void initFailSafe(final Options options, final ScheduledExecutorService
 
     private RetryPolicy<MasterServerResponse> buildRetryPolicy(final Options options) {
         RetryPolicyBuilder<MasterServerResponse> builder = FailsafeBuilder.buildRetryPolicy(FailsafeOptions.class, options);
+        builder.abortOn(p -> {
+            Throwable error = Errors.unwrap(p);
+            return (error instanceof CancellationException) || (error instanceof RejectedExecutionException) || (error instanceof InterruptedException);
+        });
         return builder.build();
     }
 

protocols/valve/steam/master/src/main/java/com/ibasco/agql/protocols/valve/steam/master/handlers/MasterServerAddressDecoder.java

@@ -71,9 +71,11 @@ protected Object decodeMessage(ChannelHandlerContext ctx, AbstractRequest reques
         if (masterRequest.getCallback() != null && fullSet.add(address)) {
             partialSet.add(address);
             try {
-                masterRequest.getCallback().accept(address, envelope.recipient(), null);
+                if (context.isValid())
+                    masterRequest.getCallback().accept(address, envelope.recipient(), null);
             } catch (Exception e) {
                 error("Error thrown by the callback", e);
+                throw e;
             }
         }
         return null;