[Hermes] Handle reconnect message (#1114)

## Pull Request Etiquette

### Checklist

- [x] Tests have been added in relevant areas
- [x] Corresponding changes made to the documentation (README.adoc) <!--
(if irrelevant check the box too) -->

### Type of change
Bug fix
<!-- Choose one from "Bug fix" / "New Feature" / "Breaking change" /
"Internal change" -->

## Description
Handle the reconnect message from the Hermes endpoint, effectively
disconnecting and reconnecting the previous subscriptions elsewhere.

<!-- Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context. List any
dependencies that are required for this change. -->

Author: Rakambda

miner/src/main/java/fr/rakambda/channelpointsminer/miner/api/hermes/TwitchHermesWebSocketClient.java

@@ -11,6 +11,7 @@
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ITwitchHermesWebSocketResponse;
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.KeepAliveResponse;
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.NotificationResponse;
+import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ReconnectResponse;
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.SubscribeResponse;
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.UnsubscribeResponse;
 import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.WelcomeResponse;
@@ -89,6 +90,10 @@ public void onMessage(String messageStr){
 					subscribeRequests.remove(unsubscribeResponse.getUnsubscribeResponse().getSubscription().getId());
 				}
 				case NotificationResponse notificationResponse -> log.debug("Received Hermes notification of type {}", notificationResponse.getNotification().getClass().getSimpleName());
+				case ReconnectResponse ignored -> {
+					log.warn("Received Hermes reconnect response, TODO: see if any field is useful : {}", messageStr);
+					close(GOING_AWAY);
+				}
 				default -> {
 				}
 			}

miner/src/main/java/fr/rakambda/channelpointsminer/miner/api/hermes/TwitchHermesWebSocketPool.java

@@ -97,6 +97,7 @@ public void onWebSocketClosed(@NotNull TwitchHermesWebSocketClient client, int c
 		clients.remove(client);
 		if(code != NORMAL){
 			pendingTopics.addAll(client.getSubscribeRequests().keySet().stream().map(topics::get).filter(Objects::nonNull).toList());
+			client.getSubscribeRequests().keySet().forEach(topics::remove);
 		}
 	}
 

miner/src/main/java/fr/rakambda/channelpointsminer/miner/api/hermes/data/response/ITwitchHermesWebSocketResponse.java

@@ -23,6 +23,7 @@
 		@JsonSubTypes.Type(value = SubscribeResponse.class, name = "subscribeResponse"),
 		@JsonSubTypes.Type(value = UnsubscribeResponse.class, name = "unsubscribeResponse"),
 		@JsonSubTypes.Type(value = NotificationResponse.class, name = "notification"),
+		@JsonSubTypes.Type(value = ReconnectResponse.class, name = "reconnect"),
 })
 @ToString
 @EqualsAndHashCode

miner/src/main/java/fr/rakambda/channelpointsminer/miner/api/hermes/data/response/ReconnectResponse.java

@@ -0,0 +1,21 @@
+package fr.rakambda.channelpointsminer.miner.api.hermes.data.response;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@JsonTypeName("reconnect")
+@Getter
+@SuperBuilder
+@EqualsAndHashCode(callSuper = true)
+@AllArgsConstructor
+@ToString(callSuper = true)
+public class ReconnectResponse extends ITwitchHermesWebSocketResponse{
+	// TODO what fields are present ?
+}

miner/src/test/java/fr/rakambda/channelpointsminer/miner/api/hermes/TwitchHermesWebSocketReconnectTest.java

@@ -0,0 +1,60 @@
+package fr.rakambda.channelpointsminer.miner.api.hermes;
+
+import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ReconnectResponse;
+import fr.rakambda.channelpointsminer.miner.tests.TestUtils;
+import fr.rakambda.channelpointsminer.miner.tests.WebsocketMockServer;
+import fr.rakambda.channelpointsminer.miner.tests.WebsocketMockServerExtension;
+import io.github.artsok.RepeatedIfExceptionsTest;
+import org.awaitility.core.ConditionTimeoutException;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import java.net.URI;
+import java.time.ZonedDateTime;
+import static org.java_websocket.framing.CloseFrame.GOING_AWAY;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+@ExtendWith(WebsocketMockServerExtension.class)
+class TwitchHermesWebSocketReconnectTest{
+	private static final int MESSAGE_TIMEOUT = 15000;
+	private TwitchHermesWebSocketClient tested;
+	
+	@Mock
+	private ITwitchHermesWebSocketListener listener;
+	
+	@AfterEach
+	void tearDown(WebsocketMockServer server){
+		tested.close();
+		server.removeClients();
+	}
+	
+	@RepeatedIfExceptionsTest(repeats = 5, exceptions = ConditionTimeoutException.class)
+	void onResponse(WebsocketMockServer server) throws InterruptedException{
+		tested.connectBlocking();
+		
+		server.send(TestUtils.getAllResourceContent("api/hermes/reconnect_ok.json"));
+		
+		var expected = ReconnectResponse.builder()
+				.id("4ae4865b-cedc-4755-8335-0560b7d05341")
+				.timestamp(ZonedDateTime.parse("2025-01-02T03:04:05.123456789Z"))
+				.build();
+		verify(listener, timeout(MESSAGE_TIMEOUT)).onWebSocketMessage(expected);
+		verify(listener).onWebSocketClosed(eq(tested), eq(GOING_AWAY), anyString(), anyBoolean());
+	}
+	
+	@BeforeEach
+	void setUp(WebsocketMockServer server){
+		var uri = URI.create("ws://127.0.0.1:" + server.getPort());
+		tested = new TwitchHermesWebSocketClient(uri);
+		tested.setReuseAddr(true);
+		tested.addListener(listener);
+	}
+}

miner/src/test/resources/api/hermes/reconnect_ok.json

@@ -0,0 +1,5 @@
+{
+  "id" : "4ae4865b-cedc-4755-8335-0560b7d05341",
+  "type" : "reconnect",
+  "timestamp" : "2025-01-02T03:04:05.123456789Z"
+}