tangled
alpha
login
or
join now
stream.place
/
streamplace
Live video on the AT Protocol
74
fork
atom
overview
issues
1
pulls
pipelines
ope fix formatting
seiso.moe
9 months ago
891b8a68
ce468d15
+39
-39
2 changed files
expand all
collapse all
unified
split
pkg
api
api.go
websocket.go
+25
-25
pkg/api/api.go
···
747
747
}
748
748
749
749
func (a *StreamplaceAPI) getLimiter(ip string) *rate.Limiter {
750
750
-
a.limitersMu.Lock()
751
751
-
defer a.limitersMu.Unlock()
752
752
-
753
753
-
limiter, exists := a.limiters[ip]
754
754
-
if !exists {
755
755
-
// 5 actions per second with a burst of 3
756
756
-
limiter = rate.NewLimiter(rate.Limit(5.0), 3)
757
757
-
a.limiters[ip] = limiter
758
758
-
}
759
759
-
760
760
-
return limiter
750
750
+
a.limitersMu.Lock()
751
751
+
defer a.limitersMu.Unlock()
752
752
+
753
753
+
limiter, exists := a.limiters[ip]
754
754
+
if !exists {
755
755
+
// 5 actions per second with a burst of 3
756
756
+
limiter = rate.NewLimiter(rate.Limit(5.0), 3)
757
757
+
a.limiters[ip] = limiter
758
758
+
}
759
759
+
760
760
+
return limiter
761
761
}
762
762
763
763
func (a *StreamplaceAPI) getMsgLimiter(connID string) *rate.Limiter {
764
764
-
a.msgLimitersMu.Lock()
765
765
-
defer a.msgLimitersMu.Unlock()
766
766
-
767
767
-
limiter, exists := a.msgLimiters[connID]
768
768
-
if !exists {
769
769
-
// 10 messages per second with a burst of 20
770
770
-
limiter = rate.NewLimiter(rate.Limit(10), 20)
771
771
-
a.msgLimiters[connID] = limiter
772
772
-
}
773
773
-
774
774
-
return limiter
764
764
+
a.msgLimitersMu.Lock()
765
765
+
defer a.msgLimitersMu.Unlock()
766
766
+
767
767
+
limiter, exists := a.msgLimiters[connID]
768
768
+
if !exists {
769
769
+
// 10 messages per second with a burst of 20
770
770
+
limiter = rate.NewLimiter(rate.Limit(10), 20)
771
771
+
a.msgLimiters[connID] = limiter
772
772
+
}
773
773
+
774
774
+
return limiter
775
775
}
776
776
777
777
func (a *StreamplaceAPI) removeMsgLimiter(connID string) {
778
778
-
a.msgLimitersMu.Lock()
779
779
-
defer a.msgLimitersMu.Unlock()
780
780
-
delete(a.msgLimiters, connID)
778
778
+
a.msgLimitersMu.Lock()
779
779
+
defer a.msgLimitersMu.Unlock()
780
780
+
delete(a.msgLimiters, connID)
781
781
}
+14
-14
pkg/api/websocket.go
···
37
37
38
38
limiter := a.getLimiter(clientIP)
39
39
if !limiter.Allow() {
40
40
-
apierrors.WriteHTTPTooManyRequests(w, "rate limit")
41
41
-
return
40
40
+
apierrors.WriteHTTPTooManyRequests(w, "rate limit")
41
41
+
return
42
42
}
43
43
44
44
uu, _ := uuid.NewV7()
···
69
69
70
70
msgLimiter := a.getMsgLimiter(connID)
71
71
defer a.removeMsgLimiter(connID)
72
72
-
72
72
+
73
73
initialBurst := make(chan any, 200)
74
74
err = conn.SetReadDeadline(time.Now().Add(30 * time.Second))
75
75
if err != nil {
···
227
227
if !r.OK() {
228
228
log.Error(ctx, "rate limit exceeded, message rejected")
229
229
230
230
-
errorMsg := map[string]string{"error": "rate limit exceeded"}
231
231
-
errorBytes, _ := json.Marshal(errorMsg)
232
232
-
conn.WriteMessage(websocket.TextMessage, errorBytes)
233
233
-
234
234
-
continue
230
230
+
errorMsg := map[string]string{"error": "rate limit exceeded"}
231
231
+
errorBytes, _ := json.Marshal(errorMsg)
232
232
+
conn.WriteMessage(websocket.TextMessage, errorBytes)
233
233
+
234
234
+
continue
235
235
}
236
236
237
237
-
// wait for rate limit delay if there is one
237
237
+
// wait for rate limit delay if there is one
238
238
delay := r.Delay()
239
239
if delay > 0 {
240
240
-
select {
241
241
-
case <-time.After(delay):
242
242
-
case <-ctx.Done():
243
243
-
return
244
244
-
}
240
240
+
select {
241
241
+
case <-time.After(delay):
242
242
+
case <-ctx.Done():
243
243
+
return
244
244
+
}
245
245
}
246
246
247
247
messageType, message, err := conn.ReadMessage()