Repository navigation
Expand file tree
/
Copy pathhttpclient.go
More file actions
146 lines (127 loc) · 3.34 KB
/
Copy pathhttpclient.go
File metadata and controls
146 lines (127 loc) · 3.34 KB
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
137
138
139
140
141
142
143
144
145
146
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package target
import (
"encoding/json"
"errors"
"net/http"
"sync/atomic"
"time"
"github.com/minio/minio/pkg/event"
xnet "github.com/minio/minio/pkg/net"
"github.com/skyrings/skyring-common/tools/uuid"
)
// HTTPClientTarget - HTTP client target.
type HTTPClientTarget struct {
id event.TargetID
w http.ResponseWriter
eventCh chan []byte
DoneCh chan struct{}
stopCh chan struct{}
isStopped uint32
isRunning uint32
}
// ID - returns target ID.
func (target HTTPClientTarget) ID() event.TargetID {
return target.id
}
func (target *HTTPClientTarget) start() {
go func() {
defer func() {
atomic.AddUint32(&target.isRunning, 1)
// Close DoneCh to indicate we are done.
close(target.DoneCh)
}()
write := func(event []byte) error {
if _, err := target.w.Write(event); err != nil {
return err
}
target.w.(http.Flusher).Flush()
return nil
}
keepAliveTicker := time.NewTicker(500 * time.Millisecond)
defer keepAliveTicker.Stop()
for {
select {
case <-target.stopCh:
// We are asked to stop.
return
case event, ok := <-target.eventCh:
if !ok {
// Got read error. Exit the goroutine.
return
}
if err := write(event); err != nil {
// Got write error to the client. Exit the goroutine.
return
}
case <-keepAliveTicker.C:
if err := write([]byte(" ")); err != nil {
// Got write error to the client. Exit the goroutine.
return
}
}
}
}()
}
// Send - sends event to HTTP client.
func (target *HTTPClientTarget) Send(eventData event.Event) error {
if atomic.LoadUint32(&target.isRunning) != 0 {
return errors.New("closed http connection")
}
data, err := json.Marshal(struct{ Records []event.Event }{[]event.Event{eventData}})
if err != nil {
return err
}
data = append(data, byte('\n'))
select {
case target.eventCh <- data:
return nil
case <-target.DoneCh:
return errors.New("error in sending event")
}
}
// Close - closes underneath goroutine.
func (target *HTTPClientTarget) Close() error {
atomic.AddUint32(&target.isStopped, 1)
if atomic.LoadUint32(&target.isStopped) == 1 {
close(target.stopCh)
}
return nil
}
func getNewUUID() (string, error) {
uuid, err := uuid.New()
if err != nil {
return "", err
}
return uuid.String(), nil
}
// NewHTTPClientTarget - creates new HTTP client target.
func NewHTTPClientTarget(host xnet.Host, w http.ResponseWriter) (*HTTPClientTarget, error) {
uuid, err := getNewUUID()
if err != nil {
return nil, err
}
c := &HTTPClientTarget{
id: event.TargetID{ID: "httpclient" + "+" + uuid + "+" + host.Name, Name: host.Port.String()},
w: w,
eventCh: make(chan []byte),
DoneCh: make(chan struct{}),
stopCh: make(chan struct{}),
}
c.start()
return c, nil
}