Added Reconnection for channels sub

This commit is contained in:
ham1255
2020-05-10 23:10:45 +04:00
parent d82f7f4bd8
commit e634985b87
5 changed files with 46 additions and 56 deletions

View File

@@ -67,6 +67,7 @@ public class AddonPlugin extends JavaPlugin {
this.getConfig().getBoolean("Redis.useSSL"));
redisSub = new RedisSub(this, jedisPool.getResource(), this.getConfig().getStringList("Channels"));
service = Executors.newFixedThreadPool(this.getConfig().getInt("Redis.Threads"));
service.execute(redisSub);
Bukkit.getLogger().info("[Govindas limework Addon] was enabled!");
}
@@ -74,7 +75,7 @@ public class AddonPlugin extends JavaPlugin {
@Override
public void onDisable(){
redisSub.unSubAndCloseConnection();
redisSub.shutdown();
service.shutdown();
try { service.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); } catch (InterruptedException e) { e.printStackTrace(); }
jedisPool.close();

View File

@@ -1,29 +0,0 @@
package net.limework.skLimework.DoNotUse;
import org.json.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
public class Tester {
public static void main(String[] args) throws InterruptedException {
JedisPoolConfig jconfig = new JedisPoolConfig();
jconfig.setMaxTotal(100);
jconfig.setMaxIdle(100);
jconfig.setMinIdle(1);
JedisPool d = new JedisPool(jconfig, "192.168.0.112", 6379, 400, "yHy0d2zdBlRmaSPj3CiBwEv5V3XxBTLTrCsGW7ntBnzhfxPxXJS6Q1aTtR6DSfAtCZr2VxWnsungXHTcF94a4bsWEpGAvjL6XMU");
Jedis dd = d.getResource();
JSONObject J = new JSONObject();
J.put("Message", "something::something::something");
int x = 0;
while (true) {
x++;
System.out.println(x);
dd.publish("fs", J.toString() );
if (x == 1000) break;
Thread.sleep(1000);
}
}
}

View File

@@ -10,27 +10,49 @@ import redis.clients.jedis.exceptions.JedisConnectionException;
import java.util.List;
public class RedisSub extends JedisPubSub {
public class RedisSub extends JedisPubSub implements Runnable{
private AddonPlugin plugin;
private Jedis j;
private String[] channels;
public RedisSub(AddonPlugin plugin, Jedis j, List<String> channels) {
this.plugin = plugin;
this.j = j;
String[] ss = channels.toArray(new String[0]);
plugin.getServer().getScheduler().runTaskAsynchronously(plugin,
() -> {
try{
this.j.subscribe(this, ss);
} catch (JedisConnectionException je){
this.unSubAndCloseConnection();
Bukkit.broadcastMessage("Redis Went down!");
}
});
this.channels = channels.toArray(new String[0]);
}
@Override
public void run(){
try{
this.j.subscribe(this, channels);
} catch (JedisConnectionException je){
plugin.getLogger().warning("Lost connection to redis!");
newJedis();
}
}
private void newJedis() {
this.unsubscribe();
this.j.close();
while (true){
try {
plugin.getLogger().info("reconnecting to Redis!");
this.j = plugin.getJedisPool().getResource();
plugin.getLogger().info("Connected!");
break;
}catch (JedisConnectionException e){
plugin.getLogger().warning("reconnecting to Redis has Failed! retrying in 4 seconds!");
try { Thread.sleep(4000);}catch (InterruptedException ignored){}
}
}
plugin.getJedisExecutionService().execute(this);
}
@Override
public void onMessage(String channel, String message) {
try {
@@ -44,8 +66,9 @@ public class RedisSub extends JedisPubSub {
}
public void unSubAndCloseConnection(){
public void shutdown(){
this.unsubscribe();
j.close();
}
}
}

View File

@@ -35,11 +35,7 @@ public class EffSendMessage extends Effect {
return;
}
plugin.getJedisExecutionService().execute(() -> {
Jedis j;
try {j = plugin.getJedisPool().getResource();}catch (JedisConnectionException e){
Bukkit.broadcastMessage("Redis is down!!! dont send messages");
return;
}
Jedis j = plugin.getJedisPool().getResource();
JSONObject json = new JSONObject();
try {
json.put("Message", message);
@@ -63,4 +59,5 @@ public class EffSendMessage extends Effect {
this.message = (Expression<String>) expressions[1];
return true;
}
}
}