Add hosted Control, full configs, and plugin-aware quick setups

README.md

@@ -2,6 +2,8 @@
 Plugin on SpigotMC
 https://www.spigotmc.org/resources/votingplugin.15358/
 
+Development documentation: [optional VotingPlugin Control discovery connector](docs/control-connector.md).
+
 ### Maven:
 
     <repository>

VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java

@@ -14,6 +14,9 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
 
 import org.bukkit.Bukkit;
 import org.bukkit.Material;
@@ -46,14 +49,19 @@
 import com.bencodez.advancedcore.api.user.AdvancedCoreUser;
 import com.bencodez.advancedcore.api.user.UserStartup;
 import com.bencodez.simpleapi.file.YMLConfig;
+import com.bencodez.simpleapi.servercomm.global.GlobalMessageHandler;
+import com.bencodez.simpleapi.servercomm.pluginmessage.PluginMessageHandler;
+import com.bencodez.simpleapi.skull.SkullCache;
 import com.bencodez.simpleapi.sql.mysql.config.MysqlConfigSpigot;
 import com.bencodez.simpleapi.time.ParsedDuration;
 import com.bencodez.simpleapi.updater.Updater;
 import com.bencodez.votingplugin.broadcast.BroadcastHandler;
 import com.bencodez.votingplugin.backendproxy.BackendProxyHandler;
+import com.bencodez.votingplugin.backendproxy.cache.ProcessedVoteCache;
 import com.bencodez.votingplugin.backgroundtask.VotingPluginBackgroundTask;
 import com.bencodez.votingplugin.backendproxy.BackendProxyRewardRegistrar;
 import com.bencodez.votingplugin.broadcast.BroadcastSettings;
+import com.bencodez.votingplugin.proxy.BungeeMethod;
 import com.bencodez.votingplugin.commands.CommandLoader;
 import com.bencodez.votingplugin.commands.executers.CommandAdminVote;
 import com.bencodez.votingplugin.commands.executers.CommandVote;
@@ -69,6 +77,7 @@
 import com.bencodez.votingplugin.config.SpecialRewardsConfig;
 import com.bencodez.votingplugin.config.VotingPluginConfigHealth;
 import com.bencodez.votingplugin.cooldown.CoolDownCheck;
+import com.bencodez.votingplugin.control.BackendControlConnector;
 import com.bencodez.votingplugin.data.ServerData;
 import com.bencodez.votingplugin.discord.DiscordHandler;
 import com.bencodez.votingplugin.listeners.BlockBreak;
@@ -147,6 +156,10 @@ public class VotingPluginMain extends AdvancedCorePlugin {
 
 	@Getter
 	private BackendProxyHandler backendProxyHandler;
+	private final ProcessedVoteCache backendProcessedVoteCache = new ProcessedVoteCache();
+	private final AtomicReference<GlobalMessageHandler> backendPluginMessageTarget = new AtomicReference<>();
+	private PluginMessageHandler backendPluginMessageRelay;
+	private BackendControlConnector backendControlConnector;
 
 	@Getter
 	private BungeeSettings bungeeSettings;
@@ -429,8 +442,20 @@ public boolean isVoteSite(String voteSite) {
 	}
 
 	private void loadBungeeHandler() {
-		backendProxyHandler = new BackendProxyHandler(this);
-		backendProxyHandler.load();
+		BackendProxyHandler candidate = new BackendProxyHandler(this, backendProcessedVoteCache);
+		try {
+			candidate.load();
+			backendProxyHandler = candidate;
+		} catch (RuntimeException failure) {
+			try {
+				candidate.close();
+			} catch (RuntimeException closeFailure) {
+				failure.addSuppressed(closeFailure);
+			}
+			backendProxyHandler = null;
+			getLogger().warning("Backend proxy transport was not started; it will be retried on reload: "
+					+ failure.getMessage());
+		}
 
 		if (getOptions().getServer().equalsIgnoreCase("PleaseSet")) {
 			getLogger().warning("Bungeecoord is true and server name is not set, bungeecoord features may not work");
@@ -715,6 +740,79 @@ public void run() {
 			}, 10);
 		}
 
+		startBackendControlConnector();
+
+	}
+
+	private void startBackendControlConnector() {
+		try {
+			backendControlConnector = BackendControlConnector.create(this);
+			if (backendControlConnector != null) backendControlConnector.start();
+		} catch (Exception e) {
+			backendControlConnector = null;
+			getLogger().warning("[Control] Bukkit configuration connector was not started: " + e.getMessage());
+		}
+	}
+
+	/** Refreshes the optional Control connector after its Config.yml settings were applied. */
+	public synchronized void restartBackendControlConnector() {
+		BackendControlConnector connector = backendControlConnector;
+		if (connector != null && connector.hasPendingResults()) {
+			getLogger().warning("[Control] Keeping the previous connector until its applied result is acknowledged");
+			return;
+		}
+		BackendControlConnector replacement;
+		try {
+			replacement = BackendControlConnector.create(this);
+		} catch (Exception e) {
+			replacement = null;
+			getLogger().warning("[Control] Bukkit configuration connector was not restarted: " + e.getMessage());
+		}
+		if (connector != null) connector.close();
+		if (backendControlConnector == connector) backendControlConnector = replacement;
+		if (replacement != null) replacement.start();
+	}
+
+	/** Recreates proxy transports after Control applies BungeeSettings.yml. */
+	public synchronized void restartBackendProxyHandler() {
+		BackendProxyHandler previous = backendProxyHandler;
+		if (!bungeeSettings.isUseBungeecoord()) {
+			backendProxyHandler = null;
+			if (previous != null) previous.close();
+			return;
+		}
+		BungeeMethod replacementMethod = BungeeMethod.getByName(bungeeSettings.getBungeeMethod());
+		if (previous != null) previous.prepareForReplacement(replacementMethod);
+		BackendProxyHandler replacement = new BackendProxyHandler(this, backendProcessedVoteCache);
+		try {
+			replacement.load();
+			replacement.validateTransport();
+			if (previous != null) previous.completeRedisHandoff(replacement);
+		} catch (RuntimeException failure) {
+			replacement.close();
+			throw failure;
+		}
+		backendProxyHandler = replacement;
+		if (previous != null) previous.close();
+	}
+
+	/** Keeps one plugin-message listener for the plugin lifetime and atomically swaps its active backend handler. */
+	public synchronized void activateBackendPluginMessageHandler(GlobalMessageHandler target) {
+		backendPluginMessageTarget.set(target);
+		if (backendPluginMessageRelay == null) {
+			backendPluginMessageRelay = new PluginMessageHandler() {
+				@Override
+				public void onReceive(com.bencodez.simpleapi.servercomm.codec.JsonEnvelope envelope) {
+					GlobalMessageHandler current = backendPluginMessageTarget.get();
+					if (current != null) current.onMessage(envelope);
+				}
+			};
+			getPluginMessaging().add(backendPluginMessageRelay);
+		}
+	}
+
+	public void deactivateBackendPluginMessageHandler(GlobalMessageHandler target) {
+		backendPluginMessageTarget.compareAndSet(target, null);
 	}
 
 	private void migrateVoteBroadcast(Config configFile) {
@@ -808,6 +906,17 @@ public void onFinish() {
 
 	@Override
 	public void onUnLoad() {
+		BackendControlConnector connector = backendControlConnector;
+		if (connector != null) {
+			try {
+				connector.close();
+			} catch (RuntimeException e) {
+				getLogger().warning("[Control] Bukkit connector did not stop cleanly; normal plugin cleanup will continue");
+				debug(e);
+			} finally {
+				if (backendControlConnector == connector) backendControlConnector = null;
+			}
+		}
 		if (getBackendProxyHandler() != null) {
 			try {
 				getBackendProxyHandler().close();
@@ -953,13 +1062,15 @@ private void reloadPlugin(boolean userStorage) {
 		reloadAdvancedCore(userStorage);
 
 		if (bungeeSettings.isUseBungeecoord()) {
-			if (getBackendProxyHandler() == null) {
+			BackendProxyHandler handler = getBackendProxyHandler();
+			if (handler == null) {
 				loadBungeeHandler();
+				handler = getBackendProxyHandler();
 			} else {
-				getBackendProxyHandler().reloadPresenceReporting();
+				handler.reloadPresenceReporting();
 			}
-			if (userStorage) {
-				getBackendProxyHandler().loadGlobalMysql();
+			if (userStorage && handler != null) {
+				handler.loadGlobalMysql();
 			}
 		} else if (getBackendProxyHandler() != null) {
 			getBackendProxyHandler().disablePresenceReporting();

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/BackendProxyHandler.java

@@ -34,7 +34,7 @@
 public class BackendProxyHandler implements Listener {
 
 	private final VotingPluginMain plugin;
-	private final ProcessedVoteCache processedVoteCache = new ProcessedVoteCache();
+	private final ProcessedVoteCache processedVoteCache;
 	private final BackendProxyTransportManager transportManager;
 	private final BackendGlobalDataSync globalDataSync;
 
@@ -48,8 +48,13 @@ public class BackendProxyHandler implements Listener {
 	private GlobalMessageHandler globalMessageHandler;
 
 	public BackendProxyHandler(VotingPluginMain plugin) {
+		this(plugin, new ProcessedVoteCache());
+	}
+
+	public BackendProxyHandler(VotingPluginMain plugin, ProcessedVoteCache processedVoteCache) {
 		this.plugin = plugin;
-		transportManager = new BackendProxyTransportManager(plugin);
+		this.processedVoteCache = java.util.Objects.requireNonNull(processedVoteCache, "processedVoteCache");
+		transportManager = new BackendProxyTransportManager(plugin, processedVoteCache);
 		globalDataSync = new BackendGlobalDataSync(plugin, this::sendEnvelope);
 	}
 
@@ -96,6 +101,28 @@ public void close() {
 		globalDataSync.close();
 	}
 
+	/** Releases a same-method subscriber/listener before its replacement starts. */
+	public void prepareForReplacement(BungeeMethod replacementMethod) {
+		if (method == replacementMethod && method != BungeeMethod.PLUGINMESSAGING && method != BungeeMethod.REDIS) {
+			transportManager.prepareForReplacement();
+		}
+	}
+
+	/** Fails a configuration apply when its selected transport did not initialize. */
+	public void validateTransport() {
+		if (method == null || globalMessageHandler == null || presenceManager == null) {
+			throw new IllegalStateException("Backend proxy handler initialization failed");
+		}
+		transportManager.validate();
+	}
+
+	/** Completes the no-loss/no-duplicate same-Redis subscriber handoff after validation. */
+	public void completeRedisHandoff(BackendProxyHandler replacement) {
+		if (method != BungeeMethod.REDIS || replacement.method != BungeeMethod.REDIS) return;
+		transportManager.closeRedisForHandoff();
+		replacement.transportManager.activateRedisAfterHandoff();
+	}
+
 	public void playerOnline(String playerName, String uuid) {
 		if (presenceManager != null) {
 			presenceManager.playerOnline(playerName, uuid);

VotingPlugin/src/main/java/com/bencodez/votingplugin/backendproxy/cache/ProcessedVoteCache.java

@@ -1,5 +1,8 @@
 package com.bencodez.votingplugin.backendproxy.cache;
 
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -12,10 +15,18 @@
 public class ProcessedVoteCache {
 
 	private static final long DEFAULT_TTL_MILLIS = TimeUnit.MINUTES.toMillis(30);
+	private static final int MAX_REDIS_DELIVERIES = 4096;
+	public static final int MAX_LEGACY_REDIS_DELIVERY_BYTES = 256 * 1024;
+	public static final int MAX_LEGACY_REDIS_TOTAL_BYTES = 4 * 1024 * 1024;
 
 	@Getter
 	private final ConcurrentHashMap<UUID, Long> processedVotes = new ConcurrentHashMap<>();
 	private final long ttlMillis;
+	private final LinkedHashMap<String, Long> processedRedisDeliveries = new LinkedHashMap<>();
+	private final LinkedHashMap<String, Integer> legacyRedisDeliveries = new LinkedHashMap<>();
+	private long legacyRedisDeliveryBytes;
+	private Object activeRedisSubscriber;
+	private Object standbyRedisSubscriber;
 
 	public ProcessedVoteCache() {
 		this(DEFAULT_TTL_MILLIS);
@@ -54,7 +65,106 @@ public boolean reserve(UUID voteId) {
 		}
 	}
 
+	/** Deduplicates one Redis envelope across overlapping subscribers during a validated handoff. */
+	public synchronized boolean reserveRedisDelivery(String deliveryId) {
+		if (deliveryId == null || !deliveryId.matches("[0-9a-fA-F-]{36}")) return true;
+		long now = System.currentTimeMillis();
+		processedRedisDeliveries.entrySet().removeIf(entry -> entry.getValue() <= now);
+		Long current = processedRedisDeliveries.get(deliveryId);
+		if (current != null && current > now) return false;
+		processedRedisDeliveries.put(deliveryId, now + ttlMillis);
+		while (processedRedisDeliveries.size() > MAX_REDIS_DELIVERIES) {
+			Iterator<Map.Entry<String, Long>> oldest = processedRedisDeliveries.entrySet().iterator();
+			if (!oldest.hasNext()) break;
+			oldest.next();
+			oldest.remove();
+		}
+		return true;
+	}
+
+	public synchronized void registerRedisSubscriber(Object subscriber) {
+		if (activeRedisSubscriber == null) {
+			activeRedisSubscriber = subscriber;
+		} else if (activeRedisSubscriber != subscriber) {
+			standbyRedisSubscriber = subscriber;
+			legacyRedisDeliveries.clear();
+			legacyRedisDeliveryBytes = 0;
+		}
+	}
+
+	/** Returns true only for the active subscriber and counts its legacy delivery during overlap. */
+	public synchronized boolean reserveLegacyRedisDelivery(Object subscriber, String signature) {
+		if (activeRedisSubscriber != subscriber) return false;
+		if (standbyRedisSubscriber != null) {
+			Integer count = legacyRedisDeliveries.get(signature);
+			if (count != null) {
+				legacyRedisDeliveries.put(signature,
+						count == Integer.MAX_VALUE ? Integer.MAX_VALUE : count + 1);
+			} else {
+				int bytes = legacyRedisDeliveryBytes(signature);
+				if (bytes <= MAX_LEGACY_REDIS_DELIVERY_BYTES
+						&& legacyRedisDeliveries.size() < MAX_REDIS_DELIVERIES
+						&& legacyRedisDeliveryBytes <= MAX_LEGACY_REDIS_TOTAL_BYTES - bytes) {
+					legacyRedisDeliveries.put(signature, 1);
+					legacyRedisDeliveryBytes += bytes;
+				}
+			}
+		}
+		return true;
+	}
+
+	public synchronized void activateRedisSubscriber(Object subscriber) {
+		if (standbyRedisSubscriber != subscriber) {
+			throw new IllegalStateException("Redis replacement subscriber is not registered");
+		}
+		activeRedisSubscriber = subscriber;
+		standbyRedisSubscriber = null;
+	}
+
+	/** Consumes one matching delivery processed by the previous active subscriber. */
+	public synchronized boolean consumeLegacyRedisDelivery(String signature) {
+		Integer count = legacyRedisDeliveries.get(signature);
+		if (count == null) return false;
+		if (count <= 1) {
+			legacyRedisDeliveries.remove(signature);
+			legacyRedisDeliveryBytes -= legacyRedisDeliveryBytes(signature);
+		} else legacyRedisDeliveries.put(signature, count - 1);
+		return true;
+	}
+
+	public synchronized void finishRedisHandoff() {
+		legacyRedisDeliveries.clear();
+		legacyRedisDeliveryBytes = 0;
+	}
+
+	public synchronized void unregisterRedisSubscriber(Object subscriber) {
+		if (standbyRedisSubscriber == subscriber) {
+			standbyRedisSubscriber = null;
+			legacyRedisDeliveries.clear();
+			legacyRedisDeliveryBytes = 0;
+		}
+		if (activeRedisSubscriber == subscriber) activeRedisSubscriber = null;
+	}
+
 	private void cleanup(long now) {
 		processedVotes.entrySet().removeIf(entry -> entry.getValue() <= now);
 	}
+
+	/** Returns an exact UTF-8 length up to the per-delivery cap, then cap + 1. */
+	public static int legacyRedisDeliveryBytes(String signature) {
+		if (signature == null) return 0;
+		int bytes = 0;
+		for (int index = 0; index < signature.length(); index++) {
+			char character = signature.charAt(index);
+			if (character <= 0x7f) bytes++;
+			else if (character <= 0x7ff) bytes += 2;
+			else if (Character.isHighSurrogate(character) && index + 1 < signature.length()
+					&& Character.isLowSurrogate(signature.charAt(index + 1))) {
+				bytes += 4;
+				index++;
+			} else bytes += 3;
+			if (bytes > MAX_LEGACY_REDIS_DELIVERY_BYTES) return MAX_LEGACY_REDIS_DELIVERY_BYTES + 1;
+		}
+		return bytes;
+	}
 }