Monorepo for Tangled
tapc: basic auth on dial, OnConnect handler
Lewis: May this revision serve well! <lewis@tangled.org>
+31
-11
+14
-3
tapc/simple_indexer.go
+14
-3
tapc/simple_indexer.go
···
3
3
import "context"
4
4
5
5
type SimpleIndexer struct {
6
-
EventHandler func(ctx context.Context, evt Event) error
7
-
ErrorHandler func(ctx context.Context, err error)
6
+
EventHandler func(ctx context.Context, evt Event) error
7
+
ErrorHandler func(ctx context.Context, err error)
8
+
ConnectHandler func(ctx context.Context)
8
9
}
9
10
10
-
var _ Handler = (*SimpleIndexer)(nil)
11
+
var (
12
+
_ Handler = (*SimpleIndexer)(nil)
13
+
_ ConnectHandler = (*SimpleIndexer)(nil)
14
+
)
11
15
12
16
func (i *SimpleIndexer) OnEvent(ctx context.Context, evt Event) error {
13
17
if i.EventHandler == nil {
···
22
26
}
23
27
i.ErrorHandler(ctx, err)
24
28
}
29
+
30
+
func (i *SimpleIndexer) OnConnect(ctx context.Context) {
31
+
if i.ConnectHandler == nil {
32
+
return
33
+
}
34
+
i.ConnectHandler(ctx)
35
+
}
+17
-8
tapc/tap.go
+17
-8
tapc/tap.go
···
5
5
import (
6
6
"bytes"
7
7
"context"
8
+
"encoding/base64"
8
9
"encoding/json"
9
10
"fmt"
10
11
"net/http"
···
19
20
type Handler interface {
20
21
OnEvent(ctx context.Context, evt Event) error
21
22
OnError(ctx context.Context, err error)
23
+
}
24
+
25
+
type ConnectHandler interface {
26
+
OnConnect(ctx context.Context)
22
27
}
23
28
24
29
type Client struct {
···
95
100
}
96
101
u.Path = "/channel"
97
102
98
-
// TODO: set auth on dial
99
-
100
103
url := u.String()
104
+
basicAuth := "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+c.AdminPassword))
101
105
102
106
var backoff int
103
107
for {
···
108
112
}
109
113
110
114
header := http.Header{
111
-
"Authorization": []string{""},
115
+
"Authorization": []string{basicAuth},
112
116
}
113
117
conn, res, err := websocket.DefaultDialer.DialContext(ctx, url, header)
114
118
if err != nil {
119
+
if backoff < 12 {
120
+
backoff++
121
+
}
115
122
l.Warn("dialing failed", "url", url, "err", err, "backoff", backoff)
116
-
time.Sleep(time.Duration(5+backoff) * time.Second)
117
-
backoff++
123
+
time.Sleep(time.Duration(5*backoff) * time.Second)
118
124
119
125
continue
120
126
}
121
-
l.Info("connected to tap service")
127
+
backoff = 0
128
+
l.Info("connected to tap service", "subscription_code", res.StatusCode)
122
129
123
-
l.Info("tap event subscription response", "code", res.StatusCode)
130
+
if ch, ok := handler.(ConnectHandler); ok {
131
+
ch.OnConnect(ctx)
132
+
}
124
133
125
134
if err = c.handleConnection(ctx, conn, handler); err != nil {
126
-
l.Warn("tap connection failed", "err", err, "backoff", backoff)
135
+
l.Warn("tap connection failed", "err", err)
127
136
}
128
137
}
129
138
}