First upload
First upload working on it still having trouble with issues not showing the username but this is a start
This commit is contained in:
minster586
@@ -0,0 +1,397 @@
|
||||
package dev.streamchat;
|
||||
|
||||
import com.google.gson.JsonObject;
|
||||
import com.google.gson.JsonParser;
|
||||
import net.minecraft.client.Minecraft;
|
||||
import net.minecraft.network.chat.Component;
|
||||
import net.minecraft.client.gui.components.toasts.SystemToast;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.WebSocket;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.EnumMap;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public final class ConnectionManager {
|
||||
private static final Logger LOGGER = Logger.getLogger("StreamChat");
|
||||
private static final String[] TWITCH_EVENTS = {
|
||||
"ChatMessage", "Follow", "Sub", "ReSub", "GiftSub", "Cheer", "Raid"
|
||||
};
|
||||
private static final String[] YOUTUBE_EVENTS = {
|
||||
"Message", "SuperChat", "SuperSticker", "MemberMilestone",
|
||||
"NewSponsor", "GiftMembershipReceived"
|
||||
};
|
||||
|
||||
private final Minecraft client;
|
||||
private final StreamChatConfig config;
|
||||
private final HttpClient httpClient = HttpClient.newBuilder()
|
||||
.connectTimeout(Duration.ofSeconds(10))
|
||||
.build();
|
||||
private final ScheduledExecutorService retryExecutor = Executors.newSingleThreadScheduledExecutor(task -> {
|
||||
Thread thread = new Thread(task, "Stream Chat reconnect");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
private final Map<Platform, WebSocket> sockets = new EnumMap<>(Platform.class);
|
||||
private final Map<Platform, String> statuses = new EnumMap<>(Platform.class);
|
||||
private final Map<Platform, Instant> lastEvents = new EnumMap<>(Platform.class);
|
||||
private final Map<Platform, Integer> retryCounts = new EnumMap<>(Platform.class);
|
||||
private final java.util.Set<Platform> scheduledRetries = java.util.EnumSet.noneOf(Platform.class);
|
||||
private boolean manuallyPaused;
|
||||
private volatile long connectionGeneration;
|
||||
private String authenticationRequestId;
|
||||
|
||||
public ConnectionManager(Minecraft client, StreamChatConfig config) {
|
||||
this.client = client;
|
||||
this.config = config;
|
||||
for (Platform platform : Platform.values()) {
|
||||
statuses.put(platform, "Disconnected");
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void start() {
|
||||
manuallyPaused = false;
|
||||
connectEnabledPlatforms();
|
||||
}
|
||||
|
||||
public synchronized void toggleConnections() {
|
||||
manuallyPaused = !manuallyPaused;
|
||||
if (manuallyPaused) {
|
||||
closeAll();
|
||||
} else {
|
||||
connectEnabledPlatforms();
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void configurationChanged() {
|
||||
if (!manuallyPaused) {
|
||||
closeAll();
|
||||
connectEnabledPlatforms();
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized String status(Platform platform) {
|
||||
return statuses.getOrDefault(platform, "Disconnected");
|
||||
}
|
||||
|
||||
public synchronized String lastEvent(Platform platform) {
|
||||
Instant timestamp = lastEvents.get(platform);
|
||||
return timestamp == null ? "Never" : timestamp.toString();
|
||||
}
|
||||
|
||||
private void connectEnabledPlatforms() {
|
||||
if (config.streamerBotEnabled && (config.twitchEnabled || config.youtubeEnabled)) {
|
||||
connect(Platform.TWITCH, config.streamerBotUrl);
|
||||
} else {
|
||||
statuses.put(Platform.TWITCH, "Disabled");
|
||||
statuses.put(Platform.YOUTUBE, "Disabled");
|
||||
}
|
||||
if (config.tikFinityEnabled && config.tiktokEnabled) {
|
||||
connect(Platform.TIKTOK, config.tikFinityUrl);
|
||||
} else {
|
||||
statuses.put(Platform.TIKTOK, "Disabled");
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void connect(Platform platform, String address) {
|
||||
long generation = connectionGeneration;
|
||||
statuses.put(platform, "Connecting");
|
||||
try {
|
||||
httpClient.newWebSocketBuilder()
|
||||
.connectTimeout(Duration.ofSeconds(10))
|
||||
.buildAsync(URI.create(address), new Listener(platform, generation))
|
||||
.whenComplete((socket, error) -> {
|
||||
synchronized (ConnectionManager.this) {
|
||||
if (generation != connectionGeneration) {
|
||||
if (socket != null) socket.sendClose(WebSocket.NORMAL_CLOSURE, "Configuration changed");
|
||||
return;
|
||||
}
|
||||
if (error != null) {
|
||||
statuses.put(platform, "Error: " + error.getMessage());
|
||||
LOGGER.log(Level.WARNING, platform + " connection failed", error);
|
||||
scheduleReconnect(platform, address, generation);
|
||||
} else {
|
||||
sockets.put(platform, socket);
|
||||
}
|
||||
}
|
||||
});
|
||||
} catch (IllegalArgumentException exception) {
|
||||
statuses.put(platform, "Error: " + exception.getMessage());
|
||||
LOGGER.log(Level.WARNING, "Invalid WebSocket address for " + platform, exception);
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void closeAll() {
|
||||
connectionGeneration++;
|
||||
for (Map.Entry<Platform, WebSocket> entry : sockets.entrySet()) {
|
||||
entry.getValue().sendClose(WebSocket.NORMAL_CLOSURE, "Disconnected by user");
|
||||
statuses.put(entry.getKey(), "Disconnected");
|
||||
if (entry.getKey() == Platform.TWITCH && config.youtubeEnabled) {
|
||||
statuses.put(Platform.YOUTUBE, "Disconnected");
|
||||
}
|
||||
}
|
||||
sockets.clear();
|
||||
scheduledRetries.clear();
|
||||
retryCounts.clear();
|
||||
}
|
||||
|
||||
private void sendStreamerBotSubscription(WebSocket socket) {
|
||||
JsonObject events = new JsonObject();
|
||||
if (config.twitchEnabled) {
|
||||
events.add("Twitch", strings(TWITCH_EVENTS));
|
||||
}
|
||||
if (config.youtubeEnabled) {
|
||||
events.add("YouTube", strings(YOUTUBE_EVENTS));
|
||||
}
|
||||
JsonObject request = new JsonObject();
|
||||
request.addProperty("request", "Subscribe");
|
||||
request.addProperty("id", UUID.randomUUID().toString());
|
||||
request.add("events", events);
|
||||
socket.sendText(request.toString(), true);
|
||||
}
|
||||
|
||||
private com.google.gson.JsonArray strings(String[] values) {
|
||||
com.google.gson.JsonArray array = new com.google.gson.JsonArray();
|
||||
for (String value : values) array.add(value);
|
||||
return array;
|
||||
}
|
||||
|
||||
private void receive(Platform socketPlatform, String raw) {
|
||||
try {
|
||||
JsonObject root = JsonParser.parseString(raw).getAsJsonObject();
|
||||
if (socketPlatform == Platform.TWITCH && handleStreamerBotControl(root)) return;
|
||||
StreamEvent event = socketPlatform == Platform.TIKTOK
|
||||
? StreamEventParser.parseTikFinity(raw)
|
||||
: StreamEventParser.parseStreamerBot(raw);
|
||||
synchronized (this) {
|
||||
lastEvents.put(event.platform(), Instant.now());
|
||||
statuses.put(event.platform(), "Connected");
|
||||
}
|
||||
|
||||
if (!isKnownEvent(event)) {
|
||||
LOGGER.warning("Unrecognized " + event.platform() + " event: " + raw);
|
||||
}
|
||||
if (!config.isEnabled(event.platform())) return;
|
||||
DisplayTarget target = config.getRoute(event);
|
||||
if (target != DisplayTarget.HIDDEN) {
|
||||
client.execute(() -> display(event, target));
|
||||
}
|
||||
} catch (RuntimeException exception) {
|
||||
LOGGER.log(Level.WARNING, "Could not parse event from " + socketPlatform + ": " + raw, exception);
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void scheduleReconnect(Platform platform, String address, long generation) {
|
||||
if (generation != connectionGeneration || manuallyPaused || !isConnectionEnabled(platform)
|
||||
|| !scheduledRetries.add(platform)) {
|
||||
return;
|
||||
}
|
||||
int attempts = retryCounts.merge(platform, 1, Integer::sum);
|
||||
long delaySeconds = Math.min(30, 1L << Math.min(attempts, 5));
|
||||
statuses.put(platform, "Reconnecting in " + delaySeconds + "s");
|
||||
retryExecutor.schedule(() -> {
|
||||
synchronized (ConnectionManager.this) {
|
||||
scheduledRetries.remove(platform);
|
||||
if (generation == connectionGeneration && !manuallyPaused && isConnectionEnabled(platform)) {
|
||||
connect(platform, address);
|
||||
}
|
||||
}
|
||||
}, delaySeconds, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private boolean isConnectionEnabled(Platform platform) {
|
||||
return platform == Platform.TWITCH
|
||||
? config.streamerBotEnabled && (config.twitchEnabled || config.youtubeEnabled)
|
||||
: config.tikFinityEnabled && config.tiktokEnabled;
|
||||
}
|
||||
|
||||
private boolean handleStreamerBotControl(JsonObject root) {
|
||||
String request = root.has("request") ? root.get("request").getAsString() : "";
|
||||
if ("Hello".equalsIgnoreCase(request)) {
|
||||
JsonObject authentication = root.has("authentication") && root.get("authentication").isJsonObject()
|
||||
? root.getAsJsonObject("authentication") : new JsonObject();
|
||||
String salt = value(authentication, "salt");
|
||||
String challenge = value(authentication, "challenge");
|
||||
if (!salt.isBlank() && !challenge.isBlank() && config.streamerBotPassword.isBlank()) {
|
||||
statuses.put(Platform.TWITCH, "Error: Streamer.bot password required");
|
||||
if (config.youtubeEnabled) statuses.put(Platform.YOUTUBE, "Error: Streamer.bot password required");
|
||||
} else if (!config.streamerBotPassword.isBlank() && !salt.isBlank() && !challenge.isBlank()) {
|
||||
JsonObject authRequest = new JsonObject();
|
||||
authRequest.addProperty("request", "Authenticate");
|
||||
authenticationRequestId = UUID.randomUUID().toString();
|
||||
authRequest.addProperty("id", authenticationRequestId);
|
||||
authRequest.addProperty("authentication", sha256(challenge + salt + config.streamerBotPassword));
|
||||
WebSocket socket = sockets.get(Platform.TWITCH);
|
||||
if (socket != null) socket.sendText(authRequest.toString(), true);
|
||||
} else {
|
||||
WebSocket socket = sockets.get(Platform.TWITCH);
|
||||
if (socket != null) sendStreamerBotSubscription(socket);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (root.has("id") && root.has("status")) {
|
||||
String status = value(root, "status");
|
||||
if ("ok".equalsIgnoreCase(status)) {
|
||||
synchronized (this) {
|
||||
statuses.put(Platform.TWITCH, "Connected");
|
||||
statuses.put(Platform.YOUTUBE, config.youtubeEnabled ? "Connected" : "Disabled");
|
||||
}
|
||||
if (authenticationRequestId != null
|
||||
&& authenticationRequestId.equals(value(root, "id"))) {
|
||||
authenticationRequestId = null;
|
||||
WebSocket socket = sockets.get(Platform.TWITCH);
|
||||
if (socket != null) sendStreamerBotSubscription(socket);
|
||||
}
|
||||
} else if (!status.isBlank()) {
|
||||
String reason = value(root, "error");
|
||||
synchronized (this) {
|
||||
statuses.put(Platform.TWITCH, "Error: " + (reason.isBlank() ? status : reason));
|
||||
if (config.youtubeEnabled) {
|
||||
statuses.put(Platform.YOUTUBE, "Error: " + (reason.isBlank() ? status : reason));
|
||||
}
|
||||
}
|
||||
LOGGER.warning("Streamer.bot request returned status: " + root);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (root.has("event")) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
private String sha256(String value) {
|
||||
try {
|
||||
byte[] digest = MessageDigest.getInstance("SHA-256")
|
||||
.digest(value.getBytes(StandardCharsets.UTF_8));
|
||||
StringBuilder result = new StringBuilder(digest.length * 2);
|
||||
for (byte item : digest) result.append(String.format(Locale.ROOT, "%02x", item));
|
||||
return result.toString();
|
||||
} catch (NoSuchAlgorithmException exception) {
|
||||
throw new IllegalStateException("SHA-256 is unavailable", exception);
|
||||
}
|
||||
}
|
||||
|
||||
private String value(JsonObject object, String key) {
|
||||
return object.has(key) && !object.get(key).isJsonNull() ? object.get(key).getAsString() : "";
|
||||
}
|
||||
|
||||
private boolean isKnownEvent(StreamEvent event) {
|
||||
if (event.platform() == Platform.TIKTOK) {
|
||||
return switch (event.type()) {
|
||||
case "chat", "gift", "share", "follow", "like", "roomUser", "subscribe" -> true;
|
||||
default -> false;
|
||||
};
|
||||
}
|
||||
return event.platform() == Platform.TWITCH
|
||||
? contains(TWITCH_EVENTS, event.type())
|
||||
: contains(YOUTUBE_EVENTS, event.type());
|
||||
}
|
||||
|
||||
private boolean contains(String[] values, String target) {
|
||||
for (String value : values) if (value.equalsIgnoreCase(target)) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
private void display(StreamEvent event, DisplayTarget target) {
|
||||
String text = config.format(event);
|
||||
Component component = Component.literal(text);
|
||||
switch (target) {
|
||||
case CHAT -> {
|
||||
if (client.gui != null) client.gui.getChat().addMessage(component);
|
||||
}
|
||||
case ACTION_BAR -> {
|
||||
if (client.player != null) client.player.displayClientMessage(component, true);
|
||||
}
|
||||
case TOAST -> SystemToast.addOrUpdate(
|
||||
client.getToasts(),
|
||||
SystemToast.SystemToastIds.TUTORIAL_HINT,
|
||||
Component.literal(event.platform() + " " + event.type()),
|
||||
Component.literal(text));
|
||||
case HIDDEN -> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private final class Listener implements WebSocket.Listener {
|
||||
private final Platform platform;
|
||||
private final long generation;
|
||||
private final StringBuilder fragments = new StringBuilder();
|
||||
|
||||
private Listener(Platform platform, long generation) {
|
||||
this.platform = platform;
|
||||
this.generation = generation;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onOpen(WebSocket webSocket) {
|
||||
synchronized (ConnectionManager.this) {
|
||||
if (generation != connectionGeneration) {
|
||||
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "Configuration changed");
|
||||
return;
|
||||
}
|
||||
sockets.put(platform, webSocket);
|
||||
boolean platformEnabled = platform == Platform.TIKTOK
|
||||
? config.tiktokEnabled : config.twitchEnabled;
|
||||
statuses.put(platform, platformEnabled ? "Connected" : "Disabled");
|
||||
retryCounts.remove(platform);
|
||||
scheduledRetries.remove(platform);
|
||||
if (platform == Platform.TWITCH) {
|
||||
statuses.put(Platform.YOUTUBE, config.youtubeEnabled ? "Connected" : "Disabled");
|
||||
}
|
||||
}
|
||||
webSocket.request(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
|
||||
if (generation != connectionGeneration) return null;
|
||||
fragments.append(data);
|
||||
if (last) {
|
||||
receive(platform, fragments.toString());
|
||||
fragments.setLength(0);
|
||||
}
|
||||
webSocket.request(1);
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
|
||||
synchronized (ConnectionManager.this) {
|
||||
if (generation == connectionGeneration) {
|
||||
sockets.remove(platform);
|
||||
statuses.put(platform, "Disconnected");
|
||||
if (platform == Platform.TWITCH && config.youtubeEnabled) {
|
||||
statuses.put(Platform.YOUTUBE, "Disconnected");
|
||||
}
|
||||
scheduleReconnect(platform, platform == Platform.TWITCH
|
||||
? config.streamerBotUrl : config.tikFinityUrl, generation);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(WebSocket webSocket, Throwable error) {
|
||||
synchronized (ConnectionManager.this) {
|
||||
if (generation == connectionGeneration) {
|
||||
statuses.put(platform, "Error: " + error.getMessage());
|
||||
scheduleReconnect(platform, platform == Platform.TWITCH
|
||||
? config.streamerBotUrl : config.tikFinityUrl, generation);
|
||||
}
|
||||
}
|
||||
LOGGER.log(Level.WARNING, platform + " WebSocket error", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user