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
|
package saslauthenticate
import (
"encoding/binary"
"io"
"github.com/segmentio/kafka-go/protocol"
)
func init() {
protocol.Register(&Request{}, &Response{})
}
type Request struct {
AuthBytes []byte `kafka:"min=v0,max=v1"`
}
func (r *Request) RawExchange(rw io.ReadWriter) (protocol.Message, error) {
if err := r.writeTo(rw); err != nil {
return nil, err
}
return r.readResp(rw)
}
func (*Request) Required(versions map[protocol.ApiKey]int16) bool {
const v0 = 0
return versions[protocol.SaslHandshake] == v0
}
func (r *Request) writeTo(w io.Writer) error {
size := len(r.AuthBytes) + 4
buf := make([]byte, size)
binary.BigEndian.PutUint32(buf[:4], uint32(len(r.AuthBytes)))
copy(buf[4:], r.AuthBytes)
_, err := w.Write(buf)
return err
}
func (r *Request) readResp(read io.Reader) (protocol.Message, error) {
var lenBuf [4]byte
if _, err := io.ReadFull(read, lenBuf[:]); err != nil {
return nil, err
}
respLen := int32(binary.BigEndian.Uint32(lenBuf[:]))
data := make([]byte, respLen)
if _, err := io.ReadFull(read, data[:]); err != nil {
return nil, err
}
return &Response{
AuthBytes: data,
}, nil
}
func (r *Request) ApiKey() protocol.ApiKey { return protocol.SaslAuthenticate }
type Response struct {
ErrorCode int16 `kafka:"min=v0,max=v1"`
ErrorMessage string `kafka:"min=v0,max=v1,nullable"`
AuthBytes []byte `kafka:"min=v0,max=v1"`
SessionLifetimeMs int64 `kafka:"min=v1,max=v1"`
}
func (r *Response) ApiKey() protocol.ApiKey { return protocol.SaslAuthenticate }
var _ protocol.RawExchanger = (*Request)(nil)
|
