Repository: EventStore/EventStoreDB-Client-Java
Pull request #245 (2023-10-27 14:42:39)

Improve error reporting in SubscriptionListener.
Changed: Improve error reporting in SubscriptionListener.

Fixes: #224
Fixes: #119 

linear[bot]: <details>
<summary><a href="https://linear.app/eventstore/issue/DB-461/java-client-subscriptionlistener-refactoring">DB-461 Java client SubscriptionListener refactoring</a></summary>
<p>

Better handling of exceptions which occur during event processing
</p>
</details>

Patch:
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
index 3188459..c8dba24 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
@@ -95,7 +95,11 @@ abstract class AbstractRegularSubscription {
                         return;
                     }
 
-                    listener.onEvent(this._subscription, ResolvedEvent.fromWire(readResp.getEvent()));
+                    try {
+                        listener.onEvent(this._subscription, ResolvedEvent.fromWire(readResp.getEvent()));
+                    } catch (Exception e) {
+                        onError(e);
+                    }
                 }
 
                 @Override
@@ -108,7 +112,8 @@ abstract class AbstractRegularSubscription {
                     if (error instanceof StatusRuntimeException) {
                         StatusRuntimeException sre = (StatusRuntimeException) error;
                         if (sre.getStatus().getCode() == Status.Code.CANCELLED) {
-                            listener.onCancelled(this._subscription);
+                            listener.onCancelled(this._subscription, null);
+                            _requestStream.onCompleted();
                             return;
                         }
 
@@ -120,7 +125,8 @@ abstract class AbstractRegularSubscription {
                         }
                     }
 
-                    listener.onError(this._subscription, error);
+                    listener.onCancelled(this._subscription, error);
+                    _requestStream.onError(Status.fromThrowable(error).asRuntimeException());
                 }
 
                 @Override
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
index f2a1a59..e840d91 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
@@ -90,7 +90,11 @@ abstract class AbstractSubscribePersistentSubscription {
 
                         int retryCount = readResp.getEvent().hasNoRetryCount() ? 0 : readResp.getEvent().getRetryCount();
 
-                        listener.onEvent(this._subscription, retryCount, ResolvedEvent.fromWire(readResp.getEvent()));
+                        try {
+                            listener.onEvent(this._subscription, retryCount, ResolvedEvent.fromWire(readResp.getEvent()));
+                        } catch (Exception e) {
+                            onError(e);
+                        }
                     }
 
                     @Override
@@ -103,7 +107,8 @@ abstract class AbstractSubscribePersistentSubscription {
                         if (error instanceof StatusRuntimeException) {
                             StatusRuntimeException sre = (StatusRuntimeException) error;
                             if (sre.getStatus().getCode() == Status.Code.CANCELLED) {
-                                listener.onCancelled(this._subscription);
+                                listener.onCancelled(this._subscription, null);
+                                _requestStream.onCompleted();
                                 return;
                             }
 
@@ -115,7 +120,8 @@ abstract class AbstractSubscribePersistentSubscription {
                             }
                         }
 
-                        listener.onError(this._subscription, error);
+                        listener.onCancelled(this._subscription, error);
+                        _requestStream.onError(Status.fromThrowable(error).asRuntimeException());
                     }
 
                     @Override
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/PersistentSubscriptionListener.java b/db-client-java/src/main/java/com/eventstore/dbclient/PersistentSubscriptionListener.java
index 095d9c6..1151822 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/PersistentSubscriptionListener.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/PersistentSubscriptionListener.java
@@ -10,24 +10,18 @@ public abstract class PersistentSubscriptionListener {
      * @param retryCount how many times the event was retried.
      * @param event a resolved event.
      */
-    public void onEvent(PersistentSubscription subscription, int retryCount, ResolvedEvent event) {
-    }
+    public void onEvent(PersistentSubscription subscription, int retryCount, ResolvedEvent event) {}
 
     /**
-     * Called when an exception was raised when processing an event.
-     * @param subscription handle to the persistent subscription.
-     * @param throwable an exception.
+     * Called when the subscription is cancelled or dropped.
+     * @param subscription handle to the subscription.
+     * @param exception an exception. null if the user initiated the cancellation.
      */
-    public void onError(PersistentSubscription subscription, Throwable throwable) {
-    }
+    public void onCancelled(PersistentSubscription subscription, Throwable exception) {}
 
     /**
-     * Called when the subscription is cancelled or dropped.
-     * @param subscription handle to the persistent subscription.
+     * Called when the subscription is confirmed by the server.
+     * @param subscription handle to the subscription.
      */
-    public void onCancelled(PersistentSubscription subscription) {
-    }
-
-    public void onConfirmation(PersistentSubscription subscription) {
-    }
+    public void onConfirmation(PersistentSubscription subscription) {}
 }
\ No newline at end of file
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java b/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
index e42deac..5cca9db 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/SubscriptionListener.java
@@ -9,25 +9,18 @@ public abstract class SubscriptionListener {
      * @param subscription handle to the subscription.
      * @param event a resolved event.
      */
-    public void onEvent(Subscription subscription, ResolvedEvent event) {
-    }
+    public void onEvent(Subscription subscription, ResolvedEvent event) {}
 
     /**
-     * Called when an exception was raised when processing an event.
+     * Called when the subscription is cancelled or dropped.
      * @param subscription handle to the subscription.
-     * @param throwable an exception.
+     * @param exception an exception. null if the user initiated the cancellation.
      */
-    public void onError(Subscription subscription, Throwable throwable) {
-    }
+    public void onCancelled(Subscription subscription, Throwable exception) {}
 
     /**
-     * Called when the subscription is cancelled or dropped.
+     * Called when the subscription is confirmed by the server.
      * @param subscription handle to the subscription.
      */
-    public void onCancelled(Subscription subscription) {
-    }
-
-    public void onConfirmation(Subscription subscription) {
-
-    }
+    public void onConfirmation(Subscription subscription) {}
 }


Test patch:
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToAllTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToAllTests.java
index 72af78d..223adcd 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToAllTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToAllTests.java
@@ -24,15 +24,7 @@ public interface SubscribeToAllTests extends Expectations {
 
         SubscriptionListener listener = new SubscriptionListener() {
             @Override
-            public void onEvent(Subscription subscription, ResolvedEvent event) {
-            }
-
-            @Override
-            public void onCancelled(Subscription subscription) {
-            }
-
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
+            public void onCancelled(Subscription subscription, Throwable throwable) {
                 if (throwable instanceof StatusRuntimeException) {
                     StatusRuntimeException statusRuntimeException = (StatusRuntimeException) throwable;
                     if (statusRuntimeException.getStatus().getCode() == Status.Code.UNAVAILABLE) {
@@ -67,12 +59,12 @@ public interface SubscribeToAllTests extends Expectations {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 fail(throwable.getMessage());
             }
         };
@@ -115,12 +107,12 @@ public interface SubscribeToAllTests extends Expectations {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 fail(throwable.getMessage());
             }
         };
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToStreamTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToStreamTests.java
index c99e6a1..a7343d5 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToStreamTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/SubscribeToStreamTests.java
@@ -22,15 +22,7 @@ public interface SubscribeToStreamTests extends Expectations {
 
         SubscriptionListener listener = new SubscriptionListener() {
             @Override
-            public void onEvent(Subscription subscription, ResolvedEvent event) {
-            }
-
-            @Override
-            public void onCancelled(Subscription subscription) {
-            }
-
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
+            public void onCancelled(Subscription subscription, Throwable throwable) {
                 if (throwable instanceof StatusRuntimeException) {
                     StatusRuntimeException statusRuntimeException = (StatusRuntimeException) throwable;
                     if (statusRuntimeException.getStatus().getCode() == Status.Code.UNAVAILABLE) {
@@ -60,12 +52,12 @@ public interface SubscribeToStreamTests extends Expectations {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 fail(throwable.getMessage());
             }
         };
@@ -96,12 +88,12 @@ public interface SubscribeToStreamTests extends Expectations {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 fail(throwable.getMessage());
             }
         }
@@ -156,12 +148,12 @@ public interface SubscribeToStreamTests extends Expectations {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 fail(throwable.getMessage());
             }
         }
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
index f747f4b..36eb120 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
@@ -40,12 +40,12 @@ public interface PersistentSubscriptionToAllWithFilterTests extends ConnectionAw
             }
 
             @Override
-            public void onCancelled(PersistentSubscription subscription) {
-                cancellation.countDown();
-            }
+            public void onCancelled(PersistentSubscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    cancellation.countDown();
+                    return;
+                }
 
-            @Override
-            public void onError(PersistentSubscription subscription, Throwable throwable) {
                 Assertions.fail(throwable.getMessage());
             }
         };
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
index 07c9ea0..1239392 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
@@ -44,13 +44,13 @@ public interface SubscribePersistentSubscriptionToStreamTests extends Connection
             }
 
             @Override
-            public void onError(PersistentSubscription subscription, Throwable throwable) {
-                result.completeExceptionally(throwable);
-            }
+            public void onCancelled(PersistentSubscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    result.complete(count);
+                    return;
+                }
 
-            @Override
-            public void onCancelled(PersistentSubscription subscription) {
-                result.complete(count);
+                result.completeExceptionally(throwable);
             }
         }).get();
 
@@ -101,13 +101,13 @@ public interface SubscribePersistentSubscriptionToStreamTests extends Connection
             }
 
             @Override
-            public void onError(PersistentSubscription subscription, Throwable throwable) {
-                result.completeExceptionally(throwable);
-            }
+            public void onCancelled(PersistentSubscription subscription, Throwable throwable) {
+                if (throwable == null) {
+                    result.complete(count);
+                    return;
+                }
 
-            @Override
-            public void onCancelled(PersistentSubscription subscription) {
-                result.complete(count);
+                result.completeExceptionally(throwable);
             }
         }).get();
 
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/samples/persistent_subscriptions/PersistentSubscriptions.java b/db-client-java/src/test/java/com/eventstore/dbclient/samples/persistent_subscriptions/PersistentSubscriptions.java
index 7a3859d..3dd5ec0 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/samples/persistent_subscriptions/PersistentSubscriptions.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/samples/persistent_subscriptions/PersistentSubscriptions.java
@@ -31,13 +31,13 @@ public class PersistentSubscriptions {
                 }
 
                 @Override
-                public void onError(PersistentSubscription subscription, Throwable throwable) {
-                    System.out.println("Subscription was dropped due to " + throwable.getMessage());
-                }
+                public void onCancelled(PersistentSubscription subscription, Throwable exception) {
+                    if (exception == null) {
+                        System.out.println("Subscription is cancelled");
+                        return;
+                    }
 
-                @Override
-                public void onCancelled(PersistentSubscription subscription) {
-                    System.out.println("Subscription is cancelled");
+                    System.out.println("Subscription was dropped due to " + exception.getMessage());
                 }
             });
         // endregion subscribe-to-persistent-subscription-to-stream
@@ -94,14 +94,13 @@ public class PersistentSubscriptions {
                     }
                 }
 
-                @Override
-                public void onError(PersistentSubscription subscription, Throwable throwable) {
-                    System.out.println("Subscription was dropped due to " + throwable.getMessage());
-                }
+                public void onCancelled(PersistentSubscription subscription, Throwable exception) {
+                    if (exception == null) {
+                        System.out.println("Subscription is cancelled");
+                        return;
+                    }
 
-                @Override
-                public void onCancelled(PersistentSubscription subscription) {
-                    System.out.println("Subscription is cancelled");
+                    System.out.println("Subscription was dropped due to " + exception.getMessage());
                 }
             });
         // endregion subscribe-to-persistent-subscription-to-all
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/samples/subscribing_to_stream/SubscribingToStream.java b/db-client-java/src/test/java/com/eventstore/dbclient/samples/subscribing_to_stream/SubscribingToStream.java
index 2f7fe1c..14603d5 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/samples/subscribing_to_stream/SubscribingToStream.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/samples/subscribing_to_stream/SubscribingToStream.java
@@ -59,8 +59,12 @@ public class SubscribingToStream {
                     }
 
                     @Override
-                    public void onError(Subscription subscription, Throwable throwable) {
-                        System.out.println("Subscription was dropped due to " + throwable.getMessage());
+                    public void onCancelled(Subscription subscription, Throwable exception) {
+                        // Subscription was dropped by the user.
+                        if (exception == null)
+                            return;
+
+                        System.out.println("Subscription was dropped due to " + exception.getMessage());
                         Resubscribe(checkpoint);
                     }
                 },
@@ -111,8 +115,12 @@ public class SubscribingToStream {
                     }
 
                     @Override
-                    public void onError(Subscription subscription, Throwable throwable) {
-                        System.out.println("Subscription was dropped due to " + throwable.getMessage());
+                    public void onCancelled(Subscription subscription, Throwable exception) {
+                        // Subscription was dropped by the user.
+                        if (exception == null)
+                            return;
+
+                        System.out.println("Subscription was dropped due to " + exception.getMessage());
                         Resubscribe(checkpoint);
                     }
                 },
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
index 31b382a..f011155 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/streams/SubscriptionTests.java
@@ -51,11 +51,10 @@ public interface SubscriptionTests extends ConnectionAware {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null)
+                    return;
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 getLogger().error("Error when running subscription", throwable);
                 Assertions.fail(throwable);
             }
@@ -114,11 +113,10 @@ public interface SubscriptionTests extends ConnectionAware {
             }
 
             @Override
-            public void onCancelled(Subscription subscription) {
-            }
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null)
+                    return;
 
-            @Override
-            public void onError(Subscription subscription, Throwable throwable) {
                 Assertions.fail(throwable);
             }
         }, SubscribeToAllOptions.get()