1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
package ingress
import (
"context"
"encoding/json"
"errors"
"io"
"net/http"
"regexp"
"strings"
"time"
"github.com/gorilla/websocket"
)
type SubscriberState string
const (
SubscriberConnecting SubscriberState = "connecting"
SubscriberRegistered SubscriberState = "registered"
SubscriberReconnecting SubscriberState = "reconnecting"
SubscriberWaiting SubscriberState = "waiting"
SubscriberClosed SubscriberState = "closed"
)
type SubscriberStatus struct {
State SubscriberState
RegisteredAt time.Time
LastError string
CloseCode int
RelayVersion string
}
const (
maxStatusError = 512
healthTimeout = 5 * time.Second
)
var urlQuery = regexp.MustCompile(`((?:https?|wss?)://[^\s"'?]+)\?[^\s"']*`)
func StatusError(err error) string {
if err == nil {
return ""
}
text := urlQuery.ReplaceAllString(err.Error(), "$1")
if len(text) > maxStatusError {
text = text[:maxStatusError]
}
return text
}
func closeCode(err error) int {
var closeErr *websocket.CloseError
if errors.As(err, &closeErr) {
return closeErr.Code
}
return 0
}
func (s *Subscriber) Status() SubscriberStatus {
s.statusMu.Lock()
defer s.statusMu.Unlock()
return s.status
}
func (s *Subscriber) Relay() string {
return httpOrigin(s.cfg.IngressProxyHost, s.cfg.TLS)
}
func (s *Subscriber) report(state SubscriberState, err error, code int, version string) {
s.statusMu.Lock()
defer s.statusMu.Unlock()
if s.status.State == SubscriberClosed {
return
}
next := s.status
if err != nil {
next.LastError = StatusError(err)
}
if code != 0 {
next.CloseCode = code
}
if version != "" {
next.RelayVersion = version
}
if next.State == state {
s.status = next
return
}
next.State = state
if state == SubscriberRegistered {
next.RegisteredAt = time.Now().UTC()
next.LastError, next.CloseCode = "", 0
}
s.status = next
if s.cfg.OnStatus != nil {
s.cfg.OnStatus(next)
}
}
func (s *Subscriber) relayVersion(ctx context.Context) string {
ctx, cancel := context.WithTimeout(ctx, healthTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.Relay()+HealthPath, nil)
if err != nil {
return ""
}
resp, err := s.cfg.HTTPClient.Do(req)
if err != nil {
return ""
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return ""
}
var health struct {
Version string `json:"version"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&health); err != nil {
return ""
}
return strings.TrimSpace(health.Version)
}
