fix: registerListener issue when stream is closed after onCall, which leads to request being dead

Author: julian-siebert

plugin/src/main/java/studio/o7/octopus/plugin/OctopusImpl.java

@@ -5,10 +5,9 @@
 import it.unimi.dsi.fastutil.objects.Object2ObjectArrayMap;
 import it.unimi.dsi.fastutil.objects.Object2ObjectMap;
 import lombok.NonNull;
-import org.bukkit.plugin.Plugin;
+import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
-import org.slf4j.Logger;
 import studio.o7.octopus.plugin.api.Octopus;
 import studio.o7.octopus.plugin.api.listener.Listener;
 import studio.o7.octopus.plugin.observer.EmptyObserver;
@@ -22,19 +21,14 @@
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicReference;
 
+@Slf4j (topic = "OctopusPlugin")
 public final class OctopusImpl implements Octopus {
     private static final EmptyObserver EMPTY_OBSERVER = new EmptyObserver();
 
     private final OctopusGrpc.OctopusStub stub = OctopusSDK.stub();
     private final OctopusGrpc.OctopusBlockingStub blockingStub = OctopusSDK.blockingStub();
     private final Object2ObjectMap<UUID, Pair<Listener, StreamObserver<ListenMessage>>> listeners = new Object2ObjectArrayMap<>();
 
-    private final Logger logger;
-
-    public OctopusImpl(Plugin plugin) {
-        logger = plugin.getSLF4JLogger();
-    }
-
     @Override
     public @NotNull Collection<Entry> get(@NonNull String keyPattern, boolean includeExpired, @Nullable Instant createdRangeStart, @Nullable Instant createdRangeEnd) {
         var builder = GetRequest.newBuilder();
@@ -58,33 +52,35 @@ public void registerListener(@NonNull Listener listener) {
         var observer = stub.listen(new StreamObserver<>() {
             @Override
             public void onNext(EventCall value) {
-                try {
-                    var request = requestRef.get();
-                    if (request == null) return;
+                var start = System.currentTimeMillis();
+                var request = requestRef.get();
+                if (request == null) return;
+
+                listener.onCall(value.getObject());
 
-                    listener.onCall(value.getObject());
+                var msg = ListenMessage.newBuilder()
+                        .setCallback(value)
+                        .build();
 
-                    var msg = ListenMessage.newBuilder()
-                            .setCallback(value)
-                            .build();
+                request = requestRef.get();
+                if (request == null) return;
 
-                    request.onNext(msg);
-                } catch (Exception e) {
-                    logger.error(e.getMessage(), e);
-                }
+                request.onNext(msg);
+                log.debug("Finished EventCall `{}` in {}ms", value.getCallId(), System.currentTimeMillis() - start);
             }
 
             @Override
             public void onError(Throwable t) {
                 requestRef.set(null);
-                logger.error("Cannot call event on listener {} with key-pattern {}", listener.getListenerUniqueId(), listener.getKeyPattern(), t);
+                log.error("Cannot call event on listener {} with key-pattern {}", listener.getListenerUniqueId(), listener.getKeyPattern(), t);
                 unregisterListener(listener);
             }
 
             @Override
             public void onCompleted() {
                 requestRef.set(null);
                 unregisterListener(listener);
+                log.debug("Completed listener `{}`", listener.getListenerUniqueId());
             }
         });
 

plugin/src/main/java/studio/o7/octopus/plugin/OctopusPlugin.java

@@ -16,7 +16,7 @@ public OctopusPlugin() {
 
     @Override
     public void onEnable() {
-        octopus = new OctopusImpl(this);
+        octopus = new OctopusImpl();
     }
 
     @Override