Skip to content

Commit 74da5cf

Browse files
RakambdaRakambda
committed
Fixes
1 parent cc94947 commit 74da5cf

84 files changed

Lines changed: 624 additions & 1497 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎miner/.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
1+
/authentication
12
/build

‎miner/src/main/java/fr/rakambda/channelpointsminer/miner/api/hermes/ITwitchHermesMessageListener.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
11
package fr.rakambda.channelpointsminer.miner.api.hermes;
22

3-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.message.IHermesMessage;
4-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.topic.Topic;
3+
import fr.rakambda.channelpointsminer.miner.api.pubsub.data.message.IPubSubMessage;
54
import org.jetbrains.annotations.NotNull;
65

7-
public interface ITwitchHermesMessageListener {
8-
void onTwitchMessage(@NotNull Topic topic, @NotNull IHermesMessage message);
6+
public interface ITwitchHermesMessageListener{
7+
void onPubSubNotification(@NotNull IPubSubMessage message);
98
}

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

Lines changed: 48 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,21 @@
11
package fr.rakambda.channelpointsminer.miner.api.hermes;
22

3-
import java.net.URI;
4-
import java.time.Instant;
5-
import java.util.Collection;
6-
import java.util.HashMap;
7-
import java.util.HashSet;
8-
import java.util.Map;
9-
import java.util.Objects;
10-
import java.util.Set;
11-
import java.util.UUID;
12-
import java.util.concurrent.ConcurrentLinkedQueue;
13-
import static org.java_websocket.framing.CloseFrame.GOING_AWAY;
143
import com.fasterxml.jackson.core.JsonProcessingException;
154
import com.fasterxml.jackson.core.type.TypeReference;
5+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.AuthenticateRequest;
166
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.ITwitchHermesWebSocketRequest;
17-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.ListenTopicRequest;
18-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.KeepAliveRequest;
19-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.UnlistenTopicRequest;
20-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.topic.Topic;
21-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.topic.Topics;
7+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.SubscribeRequest;
8+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.UnsubscribeRequest;
9+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.subscribe.PubSubSubscribeType;
2210
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.AuthenticateResponse;
2311
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ITwitchHermesWebSocketResponse;
24-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.MessageResponseHermes;
25-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.PongResponseHermes;
26-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ReconnectResponseHermes;
12+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.KeepAliveResponse;
13+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.NotificationResponse;
14+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.SubscribeResponse;
15+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.UnsubscribeResponse;
16+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.WelcomeResponse;
17+
import fr.rakambda.channelpointsminer.miner.api.passport.TwitchLogin;
18+
import fr.rakambda.channelpointsminer.miner.api.pubsub.data.request.topic.Topic;
2719
import fr.rakambda.channelpointsminer.miner.factory.TimeFactory;
2820
import fr.rakambda.channelpointsminer.miner.log.LogContext;
2921
import fr.rakambda.channelpointsminer.miner.util.json.JacksonUtils;
@@ -34,32 +26,38 @@
3426
import org.java_websocket.framing.Framedata;
3527
import org.java_websocket.handshake.ServerHandshake;
3628
import org.jetbrains.annotations.NotNull;
29+
import java.net.URI;
30+
import java.time.Instant;
31+
import java.util.Collection;
32+
import java.util.HashMap;
33+
import java.util.Map;
34+
import java.util.Objects;
35+
import java.util.Optional;
36+
import java.util.UUID;
37+
import java.util.concurrent.ConcurrentLinkedQueue;
38+
import static org.java_websocket.framing.CloseFrame.GOING_AWAY;
3739

3840
@Log4j2
3941
public class TwitchHermesWebSocketClient extends WebSocketClient{
40-
@Getter
41-
private final Set<Topics> topics;
4242
private final Collection<ITwitchHermesWebSocketListener> listeners;
4343
@Getter
4444
private final String uuid;
45-
private final Map<String, ListenTopicRequest> listenRequests;
45+
@Getter
46+
private final Map<String, SubscribeRequest> subscribeRequests;
4647

4748
@Getter
4849
private Instant lastPong;
4950

5051
public TwitchHermesWebSocketClient(@NotNull URI uri){
5152
super(uri);
5253
uuid = UUID.randomUUID().toString();
53-
listenRequests = new HashMap<>();
54+
subscribeRequests = new HashMap<>();
5455

5556
setConnectionLostTimeout(0);
56-
topics = new HashSet<>();
5757
listeners = new ConcurrentLinkedQueue<>();
5858
lastPong = Instant.EPOCH;
5959

6060
addHeader("Origin", "https://www.twitch.tv");
61-
addHeader("Sec-Websocket-Key", "g5vRgkpsUreEDo2HQn0RgQ==");
62-
addHeader("Sec-Websocket-Version", "13");
6361
}
6462

6563
@Override
@@ -77,16 +75,20 @@ public void onMessage(String messageStr){
7775
log.trace("Parsed Hermes message: {}", message);
7876

7977
switch(message){
78+
case WelcomeResponse welcomeResponse -> log.info("Received Hermes welcome with keep alive of {} seconds", welcomeResponse.getWelcome().getKeepaliveSec());
8079
case AuthenticateResponse authenticateResponse -> {
8180
if(authenticateResponse.hasError()){
8281
log.error("Received Hermes error authentication {}", authenticateResponse);
8382
close(GOING_AWAY, "Invalid credentials");
8483
}
8584
}
86-
case PongResponseHermes ignored1 -> onPong();
87-
case MessageResponseHermes messageResponse -> {
85+
case KeepAliveResponse ignored -> onPong();
86+
case SubscribeResponse subscribeResponse -> log.debug("Received Hermes subscribe response with status {}", subscribeResponse.getSubscribeResponse().getResult());
87+
case UnsubscribeResponse unsubscribeResponse -> {
88+
log.debug("Received Hermes subscribe response with status {}", unsubscribeResponse.getUnsubscribeResponse().getResult());
89+
subscribeRequests.remove(unsubscribeResponse.getUnsubscribeResponse().getSubscription().getId());
8890
}
89-
case ReconnectResponseHermes ignored -> close(GOING_AWAY);
91+
case NotificationResponse notificationResponse -> log.debug("Received Hermes notification with type {}", notificationResponse.getNotification().getType());
9092
default -> {
9193
}
9294
}
@@ -114,8 +116,8 @@ private void onPong(){
114116
lastPong = TimeFactory.now();
115117
}
116118

117-
public void ping(){
118-
send(new KeepAliveRequest());
119+
public void authenticate(@NotNull TwitchLogin twitchLogin){
120+
send(new AuthenticateRequest(twitchLogin.getAccessToken()));
119121
}
120122

121123
public void send(@NotNull ITwitchHermesWebSocketRequest request){
@@ -138,36 +140,30 @@ public void addListener(@NotNull ITwitchHermesWebSocketListener listener){
138140
listeners.add(listener);
139141
}
140142

141-
public boolean isTopicListened(@NotNull Topic topic){
142-
return topics.stream()
143-
.flatMap(t -> t.getTopics().stream())
144-
.anyMatch(t -> Objects.equals(t, topic));
143+
public boolean isPubSubTopicListened(@NotNull Topic topic){
144+
return subscribeRequests.values().stream()
145+
.map(SubscribeRequest::getSubscribe)
146+
.filter(PubSubSubscribeType.class::isInstance)
147+
.map(PubSubSubscribeType.class::cast)
148+
.anyMatch(t -> Objects.equals(t.getPubsub().getTopic(), topic.getValue()));
145149
}
146150

147-
public void listenTopic(@NotNull Topics topics){
151+
public Optional<String> listenPubSubTopic(@NotNull Topic topic){
148152
try(var ignored = LogContext.empty().withSocketId(uuid)){
149-
if(this.topics.add(topics)){
150-
var request = new ListenTopicRequest(topics);
151-
listenRequests.put(request.getNonce(), request);
152-
send(request);
153-
}
153+
var request = SubscribeRequest.pubsub(topic.getValue());
154+
subscribeRequests.put(request.getSubscribe().getId(), request);
155+
send(request);
156+
return Optional.of(request.getSubscribe().getId());
154157
}
155158
}
156159

157-
public void removeTopic(@NotNull Topic topic){
160+
public void removeSubscription(@NotNull String id){
158161
try(var ignored = LogContext.empty().withSocketId(uuid)){
159-
var topics = this.topics.stream()
160-
.filter(t -> t.getTopics().contains(topic))
161-
.toList();
162-
163-
topics.forEach(t -> {
164-
send(new UnlistenTopicRequest(t));
165-
this.topics.remove(t);
166-
});
162+
send(new UnsubscribeRequest(id));
167163
}
168164
}
169165

170-
public int getTopicCount(){
171-
return topics.stream().mapToInt(Topics::getTopicCount).sum();
166+
public int getSubscriptionCount(){
167+
return subscribeRequests.size();
172168
}
173169
}

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

Lines changed: 75 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1,46 +1,60 @@
11
package fr.rakambda.channelpointsminer.miner.api.hermes;
22

3-
import java.util.Collection;
4-
import java.util.Objects;
5-
import java.util.Queue;
6-
import java.util.concurrent.ConcurrentLinkedQueue;
7-
import static java.time.temporal.ChronoUnit.MINUTES;
8-
import static org.java_websocket.framing.CloseFrame.ABNORMAL_CLOSE;
9-
import static org.java_websocket.framing.CloseFrame.NORMAL;
10-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.topic.Topic;
11-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.request.topic.Topics;
3+
import com.fasterxml.jackson.core.type.TypeReference;
124
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.ITwitchHermesWebSocketResponse;
13-
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.MessageResponseHermes;
5+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.NotificationResponse;
6+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.UnsubscribeResponse;
7+
import fr.rakambda.channelpointsminer.miner.api.hermes.data.response.notification.PubSubNotificationType;
8+
import fr.rakambda.channelpointsminer.miner.api.passport.TwitchClient;
9+
import fr.rakambda.channelpointsminer.miner.api.passport.TwitchLogin;
10+
import fr.rakambda.channelpointsminer.miner.api.pubsub.ITwitchPubSubMessageListener;
11+
import fr.rakambda.channelpointsminer.miner.api.pubsub.data.message.IPubSubMessage;
12+
import fr.rakambda.channelpointsminer.miner.api.pubsub.data.request.topic.Topic;
1413
import fr.rakambda.channelpointsminer.miner.factory.TimeFactory;
1514
import fr.rakambda.channelpointsminer.miner.factory.TwitchWebSocketClientFactory;
15+
import fr.rakambda.channelpointsminer.miner.util.json.JacksonUtils;
1616
import lombok.extern.log4j.Log4j2;
1717
import org.java_websocket.client.WebSocketClient;
1818
import org.jetbrains.annotations.NotNull;
1919
import org.jetbrains.annotations.Nullable;
20+
import java.io.IOException;
21+
import java.util.Collection;
22+
import java.util.Map;
23+
import java.util.Objects;
24+
import java.util.Queue;
25+
import java.util.concurrent.ConcurrentHashMap;
26+
import java.util.concurrent.ConcurrentLinkedQueue;
27+
import static java.time.temporal.ChronoUnit.MINUTES;
28+
import static org.java_websocket.framing.CloseFrame.ABNORMAL_CLOSE;
29+
import static org.java_websocket.framing.CloseFrame.NORMAL;
2030

2131
@Log4j2
22-
public class TwitchHermesWebSocketPool implements AutoCloseable, ITwitchHermesWebSocketListener {
32+
public class TwitchHermesWebSocketPool implements AutoCloseable, ITwitchHermesWebSocketListener{
2333
private static final int SOCKET_TIMEOUT_MINUTES = 5;
2434

35+
private final int maxSubscriptionPerClient;
36+
private final TwitchLogin twitchLogin;
37+
2538
private final Collection<TwitchHermesWebSocketClient> clients;
2639
private final Collection<ITwitchHermesMessageListener> listeners;
27-
private final Queue<Topics> pendingTopics;
28-
private final int maxTopicPerClient;
40+
private final Collection<ITwitchPubSubMessageListener> pubSubListeners;
41+
private final Queue<Topic> pendingTopics;
42+
43+
private final Map<String, fr.rakambda.channelpointsminer.miner.api.pubsub.data.request.topic.Topic> topics;
2944

30-
public TwitchHermesWebSocketPool(int maxTopicPerClient){
31-
this.maxTopicPerClient = maxTopicPerClient;
45+
public TwitchHermesWebSocketPool(int maxSubscriptionPerClient, @NotNull TwitchLogin twitchLogin){
46+
this.maxSubscriptionPerClient = maxSubscriptionPerClient;
47+
this.twitchLogin = twitchLogin;
48+
3249
clients = new ConcurrentLinkedQueue<>();
3350
listeners = new ConcurrentLinkedQueue<>();
51+
pubSubListeners = new ConcurrentLinkedQueue<>();
3452
pendingTopics = new ConcurrentLinkedQueue<>();
53+
topics = new ConcurrentHashMap<>();
3554
}
3655

3756
public void ping(){
3857
checkStaleConnection();
39-
40-
clients.stream()
41-
.filter(WebSocketClient::isOpen)
42-
.filter(client -> !client.isClosing())
43-
.forEach(TwitchHermesWebSocketClient::ping);
4458
}
4559

4660
public void checkStaleConnection(){
@@ -49,78 +63,99 @@ public void checkStaleConnection(){
4963
.forEach(client -> client.close(ABNORMAL_CLOSE, "Timeout reached"));
5064
}
5165

52-
public void removeTopic(@NotNull Topic topic){
66+
public void removePubSubTopic(@NotNull Topic topic){
67+
var subscriptionId = topics.entrySet().stream().filter(e -> Objects.equals(e.getValue(), topic)).findFirst();
68+
if(subscriptionId.isEmpty()){
69+
return;
70+
}
5371
clients.stream()
54-
.filter(client -> client.isTopicListened(topic))
55-
.forEach(client -> client.removeTopic(topic));
72+
.filter(client -> client.isPubSubTopicListened(topic))
73+
.forEach(client -> client.removeSubscription(subscriptionId.get().getKey()));
5674
}
5775

5876
public void addListener(@NotNull ITwitchHermesMessageListener listener){
5977
listeners.add(listener);
6078
}
6179

80+
public void addPubSubListener(@NotNull ITwitchPubSubMessageListener listener){
81+
pubSubListeners.add(listener);
82+
}
83+
6284
@Override
6385
public void onWebSocketMessage(@NotNull ITwitchHermesWebSocketResponse response){
64-
if(response instanceof MessageResponseHermes m){
65-
var topic = m.getData().getTopic();
66-
var message = m.getData().getMessage();
67-
listeners.forEach(l -> l.onTwitchMessage(topic, message));
86+
if(response instanceof UnsubscribeResponse u){
87+
topics.remove(u.getUnsubscribeResponse().getSubscription().getId());
88+
}
89+
if(response instanceof NotificationResponse n){
90+
if(n.getNotification() instanceof PubSubNotificationType t){
91+
try{
92+
var topic = topics.get(n.getNotification().getSubscription().getId());
93+
var message = JacksonUtils.read(t.getPubsub(), new TypeReference<IPubSubMessage>(){});
94+
pubSubListeners.forEach(l -> l.onTwitchMessage(topic, message));
95+
}
96+
catch(IOException e){
97+
log.error("Failed to parse PubSub notification from Hermes {}", t.getPubsub(), e);
98+
}
99+
}
68100
}
69101
}
70102

71103
@Override
72104
public void onWebSocketClosed(@NotNull TwitchHermesWebSocketClient client, int code, @Nullable String reason, boolean remote){
73105
clients.remove(client);
74106
if(code != NORMAL){
75-
pendingTopics.addAll(client.getTopics());
107+
pendingTopics.addAll(client.getSubscribeRequests().keySet().stream().map(topics::get).filter(Objects::nonNull).toList());
76108
}
77109
}
78110

79-
public void listenPendingTopics(){
111+
public void listenPendingPubSubTopics(){
80112
try{
81-
Topics topic;
113+
Topic topic;
82114
while(Objects.nonNull(topic = pendingTopics.poll())){
83115
listenTopic(topic);
84116
}
85117
}
86118
catch(RuntimeException e){
87-
log.error("Failed to join pending chats", e);
119+
log.error("Failed to join pending subscriptions", e);
88120
}
89121
}
90122

91-
public void listenTopic(@NotNull Topics topics){
92-
var isListened = topics.getTopics().stream().anyMatch(this::isTopicListened);
93-
if(isListened){
123+
public void listenTopic(@NotNull Topic topic){
124+
if(isTopicListened(topic)){
94125
log.debug("Topic {} is already being listened", topics);
95126
return;
96127
}
97128

98129
try{
99-
getAvailableClient().listenTopic(topics);
130+
getAvailableClient().listenPubSubTopic(topic).ifPresent(subscriptionId -> topics.put(subscriptionId, topic));
100131
}
101132
catch(RuntimeException e){
102-
pendingTopics.add(topics);
133+
pendingTopics.add(topic);
103134
throw e;
104135
}
105136
}
106137

107138
private boolean isTopicListened(@NotNull Topic topic){
108-
return clients.stream().anyMatch(client -> client.isTopicListened(topic));
139+
return clients.stream().anyMatch(client -> client.isPubSubTopicListened(topic));
109140
}
110141

111142
@NotNull
112143
private TwitchHermesWebSocketClient getAvailableClient(){
113144
return clients.stream()
114145
.filter(client -> !client.isClosing() && !client.isClosed())
115-
.filter(client -> client.getTopicCount() < maxTopicPerClient)
146+
.filter(client -> client.getSubscriptionCount() < maxSubscriptionPerClient)
116147
.findAny()
117-
.orElseGet(this::createNewClient);
148+
.orElseGet(() -> {
149+
var client = createNewClient();
150+
client.authenticate(twitchLogin);
151+
return client;
152+
});
118153
}
119154

120155
@NotNull
121156
public TwitchHermesWebSocketClient createNewClient(){
122157
try{
123-
var client = TwitchWebSocketClientFactory.createHermesClient();
158+
var client = TwitchWebSocketClientFactory.createHermesClient(TwitchClient.WEB);
124159
log.debug("Created Hermes WebSocket client with uuid {}", client.getUuid());
125160
client.addListener(this);
126161
client.connectBlocking();

0 commit comments

Comments
 (0)