-
Notifications
You must be signed in to change notification settings - Fork 2.1k
/
fake_loggerevent_streamingclient.go
153 lines (131 loc) · 4.48 KB
/
fake_loggerevent_streamingclient.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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
/*
Copyright 2019 The Vitess Authors.
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 fakevtctlclient
import (
"fmt"
"io"
"strings"
"sync"
"time"
"vitess.io/vitess/go/vt/logutil"
logutilpb "vitess.io/vitess/go/vt/proto/logutil"
)
// FakeLoggerEventStreamingClient is the base for the fakes for the vtctlclient and vtworkerclient.
// It allows to register a (multi-)line string for a given command and return the result as channel which streams it back.
type FakeLoggerEventStreamingClient struct {
results map[string]*result
// mu guards all fields of the structs.
mu sync.Mutex
}
// NewFakeLoggerEventStreamingClient creates a new fake.
func NewFakeLoggerEventStreamingClient() *FakeLoggerEventStreamingClient {
return &FakeLoggerEventStreamingClient{results: make(map[string]*result)}
}
// generateKey returns a map key for a []string.
// ([]string is not supported as map key.)
func generateKey(args []string) string {
return strings.Join(args, " ")
}
// result contains the result the fake should respond for a given command.
type result struct {
output string
err error
// count is the number of times this result is registered for the same
// command. With each stream of this result, count will be decreased by one.
count int
// addr optionally specifies which server address is expected from the client.
addr string
}
func (r1 result) Equals(r2 result) bool {
return r1.output == r2.output &&
((r1.err == nil && r2.err == nil) ||
(r1.err != nil && r2.err != nil && r1.err.Error() == r2.err.Error()))
}
// RegisterResult registers for a given command (args) the result which the fake should return.
// Once the result was returned, it will be automatically deregistered.
func (f *FakeLoggerEventStreamingClient) RegisterResult(args []string, output string, err error) error {
return f.RegisterResultForAddr("" /* addr */, args, output, err)
}
// RegisterResultForAddr is identical to RegisterResult but also expects that
// the client did dial "addr" as server address.
func (f *FakeLoggerEventStreamingClient) RegisterResultForAddr(addr string, args []string, output string, err error) error {
f.mu.Lock()
defer f.mu.Unlock()
k := generateKey(args)
v := result{output, err, 1, addr}
if result, ok := f.results[k]; ok {
if result.Equals(v) {
result.count++
return nil
}
return fmt.Errorf("a different result (%v) is already registered for command: %v", result, args)
}
f.results[k] = &v
return nil
}
// RegisteredCommands returns a list of commands which are currently registered.
// This is useful to check that all registered results have been consumed.
func (f *FakeLoggerEventStreamingClient) RegisteredCommands() []string {
f.mu.Lock()
defer f.mu.Unlock()
var commands []string
for k := range f.results {
commands = append(commands, k)
}
return commands
}
type streamResultAdapter struct {
lines []string
index int
err error
}
func (s *streamResultAdapter) Recv() (*logutilpb.Event, error) {
if s.index < len(s.lines) {
result := &logutilpb.Event{
Time: logutil.TimeToProto(time.Now()),
Level: logutilpb.Level_CONSOLE,
File: "fakevtctlclient",
Line: -1,
Value: s.lines[s.index],
}
s.index++
return result, nil
}
if s.err == nil {
return nil, io.EOF
}
return nil, s.err
}
// StreamResult returns an EventStream which streams back a registered result as logging events.
// "addr" is the server address which the client dialed and may be empty.
func (f *FakeLoggerEventStreamingClient) StreamResult(addr string, args []string) (logutil.EventStream, error) {
f.mu.Lock()
defer f.mu.Unlock()
k := generateKey(args)
result, ok := f.results[k]
if !ok {
return nil, fmt.Errorf("no response was registered for args: %v", args)
}
if result.addr != "" && addr != result.addr {
return nil, fmt.Errorf("client sent request to wrong server address. got: %v want: %v", addr, result.addr)
}
result.count--
if result.count == 0 {
delete(f.results, k)
}
return &streamResultAdapter{
lines: strings.Split(result.output, "\n"),
index: 0,
err: result.err,
}, nil
}