mongodb document subscriber

src/main/java/net/krakatoa/proxy/listener/PlayerDisconnectListener.java

@@ -13,6 +13,9 @@ import net.md_5.bungee.api.plugin.Listener;
 import net.md_5.bungee.event.EventHandler;
 import org.bson.Document;
 import org.bson.conversions.Bson;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 import org.redisson.api.RMap;
 
 public class PlayerDisconnectListener implements Listener {
@@ -30,27 +33,34 @@ public class PlayerDisconnectListener implements Listener {
         .getMap("player:" + uuid);
 
     // mongo
-    Document document = (Document) ProxySystem.getInstance().getMongoManager()
+    Publisher<Document> document = ProxySystem.getInstance().getMongoManager()
         .getPlayersCollection()
         .find(Filters.eq("uuid", uuid)).first();
 
-    if (document != null) {
+    document.subscribe(new Subscriber<Document>() {
-      List<Bson> updatesList = new ArrayList<>();
+      @Override
+      public void onSubscribe(Subscription subscription) {
+        subscription.request(1);
+      }
 
-      map.forEach((key, value) -> {
+      @Override
-        if (!Objects.equals(key, "_id") && !Objects.equals(key, "uuid")) {
+      public void onNext(Document document) {
-          updatesList.add(Updates.set(key, value));
+        List<Bson> updatesList = new ArrayList<>();
-        }
-      });
 
-      Bson updates = Updates.combine(updatesList);
+        map.forEach((key, value) -> {
+          if (!Objects.equals(key, "_id") && !Objects.equals(key, "uuid")) {
+            updatesList.add(Updates.set(key, value));
+          }
+        });
 
-      UpdateOptions options = new UpdateOptions().upsert(true);
+        Bson updates = Updates.combine(updatesList);
 
-      System.out.println("doc: " + document.keySet());
+        UpdateOptions options = new UpdateOptions().upsert(true);
 
-      ProxySystem.getInstance().getMongoManager().getPlayersCollection()
+        System.out.println("doc: " + document.keySet());
-          .updateOne(Filters.eq("uuid", uuid), updates);
+
+        ProxySystem.getInstance().getMongoManager().getPlayersCollection()
+            .updateOne(Filters.eq("uuid", uuid), updates);
 /*
             ProxySystem.getInstance().getMongoManager().getPlayersCollection()
                 .updateOne(document, updates, options, (result, t) -> {
@@ -61,9 +71,18 @@ public class PlayerDisconnectListener implements Listener {
                     System.out.println("Upserted id: " + result.getUpsertedId());
                   }
                 }); */
-    } else {
+      }
-      System.out.println("Mongo player is null on disconnect");
+
-    }
+      @Override
+      public void onError(Throwable throwable) {
+        System.out.println("Mongo player is null on disconnect");
+      }
+
+      @Override
+      public void onComplete() {
+
+      }
+    });
 
     // redis
     if (!map.isEmpty()) {

src/main/java/net/krakatoa/proxy/listener/PostLoginListener.java

@@ -35,8 +35,7 @@ public class PostLoginListener implements Listener {
     document.subscribe(new Subscriber<Document>() {
       @Override
       public void onSubscribe(Subscription subscription) {
-        //subscription.request(1);
+        subscription.request(1);
-        ProxySystem.getInstance().getLogger().info("subscripe");
       }
 
       @Override