mongodb reactive stream driver

pom.xml

@@ -90,6 +90,11 @@
             <artifactId>KraSocketClient</artifactId>
             <version>1.2-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>earth.krakatoa</groupId>
+            <artifactId>KraCore</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <repositories>

src/main/java/net/krakatoa/proxy/ProxySystem.java

@@ -5,6 +5,11 @@ import earth.krakatao.KraSocketClientConfig;
 import earth.krakatao.events.KraSocketClientEventInitiater;
 import earth.krakatao.events.KraSocketClientEventInterface;
 import earth.krakatao.protocol.KraSocketClientProtocol;
+import earth.krakatoa.core.config.MongoConfig;
+import earth.krakatoa.core.config.RedisConfig;
+import earth.krakatoa.core.mongo.MongoManager;
+import earth.krakatoa.core.redis.RedisManager;
+import earth.krakatoa.core.util.Formatter;
 import java.util.Arrays;
 import lombok.Getter;
 import lombok.SneakyThrows;
@@ -16,9 +21,6 @@ import net.krakatoa.proxy.listener.LoginListener;
 import net.krakatoa.proxy.listener.PlayerDisconnectListener;
 import net.krakatoa.proxy.listener.PostLoginListener;
 import net.krakatoa.proxy.listener.SocketClientMessageListener;
-import net.krakatoa.proxy.mongo.MongoManager;
-import net.krakatoa.proxy.redis.RedisManager;
-import net.krakatoa.proxy.util.Formatter;
 import net.md_5.bungee.api.plugin.Plugin;
 import net.md_5.bungee.api.plugin.PluginManager;
 
@@ -62,12 +64,13 @@ public class ProxySystem extends Plugin {
 
     this.kraSocketClient.getSocketClient().connect();
 
-    this.mongoManager = new MongoManager(this.configHandler.getMongodbHost(),
+    this.mongoManager = new MongoManager(
-        this.configHandler.getMongodbPort(), this.configHandler.getMongodbUsername(),
+        new MongoConfig(this.configHandler.getMongodbHost(), this.configHandler.getMongodbPort(),
-        this.configHandler.getMongodbPassword());
+            this.configHandler.getMongodbUsername(), this.configHandler.getMongodbPassword(),
-    this.mongoManager.connect(this.configHandler.getMongodbDatabase());
+            this.configHandler.getMongodbDatabase()));
 
-    this.redisManager = new RedisManager("redis://127.0.0.1:6379");
+    this.redisManager = new RedisManager(
+        new RedisConfig(this.configHandler.getRedisHost(), this.configHandler.getRedisPort()));
     this.redisManager.connect();
 
     loadListeners();
@@ -79,9 +82,8 @@ public class ProxySystem extends Plugin {
     this.kraSocketClient.getSocketClient().getConnection().close();
     this.kraSocketClient.getSocketClient().getConnection().closeConnection(1, "");
 
-    this.mongoManager.getMongoClient().close();
+    this.mongoManager.shutdown();
-
+    this.redisManager.shutdown();
-    this.redisManager.close();
 
     instance = null;
   }

src/main/java/net/krakatoa/proxy/command/KrakatoapCommand.java

@@ -33,8 +33,7 @@ public class KrakatoapCommand extends Command {
                     .getKraSocketClientProtocol()
                     .getReceivedQueueMessages().size())));
 
-        String uuid = ProxySystem.getInstance().getFormatter()
+        String uuid = ProxySystem.getInstance().getFormatter().formatUuid(player.getUniqueId());
-            .formatUuid(player.getUniqueId().toString());
 
         RMap<String, String> map = ProxySystem.getInstance().getRedisManager()
             .getRedissonClient()

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

@@ -11,6 +11,7 @@ import net.md_5.bungee.api.connection.ProxiedPlayer;
 import net.md_5.bungee.api.event.PlayerDisconnectEvent;
 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.redisson.api.RMap;
 
@@ -29,9 +30,10 @@ public class PlayerDisconnectListener implements Listener {
         .getMap("player:" + uuid);
 
     // mongo
-    ProxySystem.getInstance().getMongoManager().getPlayers()
+    Document document = (Document) ProxySystem.getInstance().getMongoManager()
-        .find(Filters.eq("uuid", uuid))
+        .getPlayersCollection()
-        .first((document, throwable) -> {
+        .find(Filters.eq("uuid", uuid)).first();
+
     if (document != null) {
       List<Bson> updatesList = new ArrayList<>();
 
@@ -47,7 +49,10 @@ public class PlayerDisconnectListener implements Listener {
 
       System.out.println("doc: " + document.keySet());
 
-            ProxySystem.getInstance().getMongoManager().getPlayers()
+      ProxySystem.getInstance().getMongoManager().getPlayersCollection()
+          .updateOne(Filters.eq("uuid", uuid), updates);
+/*
+            ProxySystem.getInstance().getMongoManager().getPlayersCollection()
                 .updateOne(document, updates, options, (result, t) -> {
                   System.out.println(result == null);
 
@@ -55,11 +60,10 @@ public class PlayerDisconnectListener implements Listener {
                     System.out.println("Modified document count: " + result.getModifiedCount());
                     System.out.println("Upserted id: " + result.getUpsertedId());
                   }
-                });
+                }); */
     } else {
       System.out.println("Mongo player is null on disconnect");
     }
-        });
 
     // redis
     if (!map.isEmpty()) {

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

@@ -37,19 +37,17 @@ public class PostLoginListener implements Listener {
     String uuid = ProxySystem.getInstance().getFormatter()
         .formatUuid(proxiedPlayer.getUniqueId().toString());
 
-    ProxySystem.getInstance().getMongoManager().getPlayers()
+    Document document = (Document) ProxySystem.getInstance().getMongoManager()
-        .find(Filters.eq("uuid", uuid))
+        .getPlayersCollection()
-        .first((document, throwable) -> {
+        .find(Filters.eq("uuid", uuid)).first();
+
     if (document == null) { // new player - create entries in databases
       System.out.println("create player in db");
       String voiceWebCode = generateVoiceWebCode(uuid);
 
       document = new Document("uuid", uuid).append("voiceWebCode", voiceWebCode);
 
-            ProxySystem.getInstance().getMongoManager().getPlayers()
+      ProxySystem.getInstance().getMongoManager().getPlayersCollection().insertOne(document);
-                .insertOne(document, (unused, throwable1) -> {
-                  System.out.println("insertOne");
-                });
 
       sendVoiceWebCodeUrl(proxiedPlayer, voiceWebCode);
 
@@ -121,9 +119,8 @@ public class PostLoginListener implements Listener {
 
           UpdateOptions options = new UpdateOptions().upsert(true);
 
-                ProxySystem.getInstance().getMongoManager().getPlayers()
+          ProxySystem.getInstance().getMongoManager().getPlayersCollection()
-                    .updateOne(finalDocument, updates, options, (updateResult, throwable1) -> {
+              .updateOne(finalDocument, updates, options);
-                    });
 
           System.out.println("voiceWebCode " + voiceWebCode);
 
@@ -140,7 +137,7 @@ public class PostLoginListener implements Listener {
               KraSocketClientProtocolDest.BACKEND.getStatus(), proxiedPlayer.getUniqueId(),
               (short) 10, "", consumer));
     }
-        });
+
 
     /*
     MessageBufferPacker packer = MessagePack.newDefaultBufferPacker();

src/main/java/net/krakatoa/proxy/mongo/MongoManager.java

@@ -1,40 +0,0 @@
-package net.krakatoa.proxy.mongo;
-
-import com.mongodb.async.client.MongoClient;
-import com.mongodb.async.client.MongoClients;
-import com.mongodb.async.client.MongoCollection;
-import com.mongodb.async.client.MongoDatabase;
-import lombok.Getter;
-import org.bson.Document;
-
-public class MongoManager {
-
-  private final String hostname;
-  private final int port;
-  private final String username;
-  private final String password;
-
-  // TODO: search for replacements for deprecated methods
-  @Getter
-  private MongoDatabase mongoDatabase;
-  @Getter
-  private MongoClient mongoClient;
-  @Getter
-  private MongoCollection<Document> players;
-
-  public MongoManager(String hostname, int port, String username, String password) {
-    this.hostname = hostname;
-    this.port = port;
-    this.username = username;
-    this.password = password;
-  }
-
-  public void connect(String database) {
-    mongoClient = MongoClients.create(
-        "mongodb://" + username + ":" + password + "@" + hostname + ":" + port + "/?authSource="
-            + database);
-
-    mongoDatabase = mongoClient.getDatabase(database);
-    players = mongoDatabase.getCollection("players");
-  }
-}

src/main/java/net/krakatoa/proxy/redis/PlayerTest.java

@@ -1,18 +0,0 @@
-package net.krakatoa.proxy.redis;
-
-import lombok.Data;
-
-@Data
-public class PlayerTest {
-
-  private String name;
-  private int money;
-  private String car;
-
-  public PlayerTest(String name, int money, String car) {
-    this.name = name;
-    this.money = money;
-    this.car = car;
-  }
-
-}

src/main/java/net/krakatoa/proxy/redis/RedisManager.java

@@ -1,37 +0,0 @@
-package net.krakatoa.proxy.redis;
-
-import lombok.Data;
-import org.redisson.Redisson;
-import org.redisson.api.RedissonClient;
-import org.redisson.config.Config;
-
-@Data
-public class RedisManager {
-
-  private final String connectionUri;
-
-  private Config config;
-  private RedissonClient redissonClient;
-
-  //private StatefulRedisConnection<String, String> connection;
-  //private RedisStringCommands<String, String> commands;
-  //private RedisKeyReactiveCommands<String, String> keyReactiveCommands;
-
-
-  public void connect() {
-    this.config = new Config();
-    //this.config.setCodec(new MsgPackJacksonCodec());
-    this.config.useSingleServer().setAddress(this.connectionUri);
-
-    this.redissonClient = Redisson.create(this.config);
-
-    //this.connection = client.connect();
-    //this.commands = connection.sync();
-    //this.keyReactiveCommands = this.connection.reactive();
-  }
-
-  public void close() {
-    this.redissonClient.shutdown();
-  }
-}
-// RedisKeyReactiveCommands

src/main/java/net/krakatoa/proxy/util/Formatter.java

@@ -1,19 +0,0 @@
-package net.krakatoa.proxy.util;
-
-import java.math.BigInteger;
-import java.util.UUID;
-
-public class Formatter {
-
-  public String formatUuid(String uuid) {
-    return uuid.replaceAll("-", "");
-  }
-
-  public UUID stringToUuid(String s) {
-    BigInteger bigInteger = new BigInteger(s.substring(0, 16), 16);
-    BigInteger bigInteger1 = new BigInteger(s.substring(16, 32), 16);
-
-    return new UUID(bigInteger.longValue(), bigInteger1.longValue());
-  }
-
-}