-
-
Notifications
You must be signed in to change notification settings - Fork 17
/
emitEvent.go
136 lines (122 loc) · 4.34 KB
/
emitEvent.go
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
133
134
135
136
package service
import (
"context"
"fmt"
"io"
netHttp "net/http"
"strconv"
"time"
"github.com/plgd-dev/hub/v2/cloud2cloud-connector/events"
"github.com/plgd-dev/hub/v2/cloud2cloud-gateway/store"
"github.com/plgd-dev/hub/v2/pkg/fsnotify"
"github.com/plgd-dev/hub/v2/pkg/log"
cmClient "github.com/plgd-dev/hub/v2/pkg/security/certManager/client"
)
type (
incrementSubscriptionSequenceNumberFunc func(ctx context.Context) (uint64, error)
emitEventFunc func(ctx context.Context, eventType events.EventType, s store.Subscription, incrementSubscriptionSequenceNumber incrementSubscriptionSequenceNumberFunc, rep interface{}) (remove bool, err error)
)
func makeEmitEventRequestBody(accept []string, rep interface{}) ([]byte, string, error) {
encoder, contentType, err := getEncoder(accept)
if err != nil {
return nil, "", fmt.Errorf("cannot get encoder: %w", err)
}
body, err := encoder(rep)
if err != nil {
return nil, "", fmt.Errorf("cannot encode data to body: %w", err)
}
return body, contentType, nil
}
func makeEmitEventRequest(ctx context.Context, eventType events.EventType, s store.Subscription, seqNum uint64, rep interface{}) (*netHttp.Request, error) {
r, w := io.Pipe()
req, err := netHttp.NewRequestWithContext(ctx, netHttp.MethodPost, s.URL, r)
if err != nil {
return nil, fmt.Errorf("cannot create post request: %w", err)
}
timestamp := time.Now()
req.Header.Set(events.EventTypeKey, string(eventType))
req.Header.Set(events.SubscriptionIDKey, s.ID)
req.Header.Set(events.SequenceNumberKey, strconv.FormatUint(seqNum, 10))
req.Header.Set(events.CorrelationIDKey, s.CorrelationID)
req.Header.Set(events.EventTimestampKey, strconv.FormatInt(timestamp.Unix(), 10))
var body []byte
if rep != nil {
var contentType string
body, contentType, err = makeEmitEventRequestBody(s.Accept, rep)
if err != nil {
return nil, fmt.Errorf("cannot create post request body: %w", err)
}
req.Header.Set(events.ContentTypeKey, contentType)
}
if len(body) > 0 {
go func() {
defer func() {
if errC := w.Close(); errC != nil {
log.Errorf("failed to close write pipe: %w", errC)
}
}()
_, err := w.Write(body)
if err != nil {
log.Errorf("cannot write data to client: %w", err)
}
}()
} else {
if errC := w.Close(); errC != nil {
log.Errorf("failed to close write pipe: %w", errC)
}
}
req.Header.Set(events.EventSignatureKey, events.CalculateEventSignature(
s.SigningSecret,
req.Header.Get(events.ContentTypeKey),
eventType,
req.Header.Get(events.SubscriptionIDKey),
seqNum,
timestamp,
body,
))
req.Header.Set("Connection", "close")
req.Close = true
return req, nil
}
func createEmitEventFunc(cfg cmClient.Config, timeout time.Duration, fileWatcher *fsnotify.Watcher, logger log.Logger) (emitEventFunc, func(), error) {
certManager, err := cmClient.New(cfg, fileWatcher, logger)
if err != nil {
return nil, nil, fmt.Errorf("cannot create cert manager: %w", err)
}
closeFunc := func() {
certManager.Close()
}
trans := netHttp.DefaultTransport.(*netHttp.Transport).Clone()
trans.TLSClientConfig = certManager.GetTLSConfig()
client := netHttp.Client{
Transport: trans,
}
emitFunc := func(ctx context.Context, eventType events.EventType, s store.Subscription, incrementSubscriptionSequenceNumber incrementSubscriptionSequenceNumberFunc, rep interface{}) (remove bool, err error) {
log.Debugf("emitEvent: %v: %+v\n rep:%v\n", eventType, s, rep)
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
seqNum, err := incrementSubscriptionSequenceNumber(ctx)
if err != nil {
return false, fmt.Errorf("cannot increment sequence number: %w", err)
}
req, err := makeEmitEventRequest(ctx, eventType, s, seqNum, rep)
if err != nil {
return false, fmt.Errorf("cannot create request: %w", err)
}
resp, err := client.Do(req)
if err != nil {
return false, fmt.Errorf("cannot post: %w", err)
}
defer func() {
if errC := resp.Body.Close(); errC != nil {
log.Errorf("failed to close response body stream: %w", errC)
}
}()
if resp.StatusCode != netHttp.StatusOK {
errBody, _ := io.ReadAll(resp.Body)
return resp.StatusCode == netHttp.StatusGone, fmt.Errorf("%v: unexpected statusCode %v: body: '%v'", s.URL, resp.StatusCode, string(errBody))
}
return eventType == events.EventType_SubscriptionCanceled, nil
}
return emitFunc, closeFunc, nil
}