VOL-1967 move api-server to separate repository
Change-Id: I21b85be74205805be15f8a85e53a903d16785671
diff --git a/vendor/github.com/Shopify/sarama/mockbroker.go b/vendor/github.com/Shopify/sarama/mockbroker.go
index 55ef1e2..4ed46a6 100644
--- a/vendor/github.com/Shopify/sarama/mockbroker.go
+++ b/vendor/github.com/Shopify/sarama/mockbroker.go
@@ -18,6 +18,8 @@
expectationTimeout = 500 * time.Millisecond
)
+type GSSApiHandlerFunc func([]byte) []byte
+
type requestHandlerFunc func(req *request) (res encoder)
// RequestNotifierFunc is invoked when a mock broker processes a request successfully
@@ -49,18 +51,19 @@
// It is not necessary to prefix message length or correlation ID to your
// response bytes, the server does that automatically as a convenience.
type MockBroker struct {
- brokerID int32
- port int32
- closing chan none
- stopper chan none
- expectations chan encoder
- listener net.Listener
- t TestReporter
- latency time.Duration
- handler requestHandlerFunc
- notifier RequestNotifierFunc
- history []RequestResponse
- lock sync.Mutex
+ brokerID int32
+ port int32
+ closing chan none
+ stopper chan none
+ expectations chan encoder
+ listener net.Listener
+ t TestReporter
+ latency time.Duration
+ handler requestHandlerFunc
+ notifier RequestNotifierFunc
+ history []RequestResponse
+ lock sync.Mutex
+ gssApiHandler GSSApiHandlerFunc
}
// RequestResponse represents a Request/Response pair processed by MockBroker.
@@ -173,6 +176,43 @@
Logger.Printf("*** mockbroker/%d: listener closed, err=%v", b.BrokerID(), err)
}
+func (b *MockBroker) SetGSSAPIHandler(handler GSSApiHandlerFunc) {
+ b.gssApiHandler = handler
+}
+
+func (b *MockBroker) readToBytes(r io.Reader) ([]byte, error) {
+ var (
+ bytesRead int
+ lengthBytes = make([]byte, 4)
+ )
+
+ if _, err := io.ReadFull(r, lengthBytes); err != nil {
+ return nil, err
+ }
+
+ bytesRead += len(lengthBytes)
+ length := int32(binary.BigEndian.Uint32(lengthBytes))
+
+ if length <= 4 || length > MaxRequestSize {
+ return nil, PacketDecodingError{fmt.Sprintf("message of length %d too large or too small", length)}
+ }
+
+ encodedReq := make([]byte, length)
+ if _, err := io.ReadFull(r, encodedReq); err != nil {
+ return nil, err
+ }
+
+ bytesRead += len(encodedReq)
+
+ fullBytes := append(lengthBytes, encodedReq...)
+
+ return fullBytes, nil
+}
+
+func (b *MockBroker) isGSSAPI(buffer []byte) bool {
+ return buffer[4] == 0x60 || bytes.Equal(buffer[4:6], []byte{0x05, 0x04})
+}
+
func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup) {
defer wg.Done()
defer func() {
@@ -192,59 +232,92 @@
}()
resHeader := make([]byte, 8)
+ var bytesWritten int
+ var bytesRead int
for {
- req, bytesRead, err := decodeRequest(conn)
+
+ buffer, err := b.readToBytes(conn)
if err != nil {
- Logger.Printf("*** mockbroker/%d/%d: invalid request: err=%+v, %+v", b.brokerID, idx, err, spew.Sdump(req))
+ Logger.Printf("*** mockbroker/%d/%d: invalid request: err=%+v, %+v", b.brokerID, idx, err, spew.Sdump(buffer))
b.serverError(err)
break
}
- if b.latency > 0 {
- time.Sleep(b.latency)
- }
+ bytesWritten = 0
+ if !b.isGSSAPI(buffer) {
- b.lock.Lock()
- res := b.handler(req)
- b.history = append(b.history, RequestResponse{req.body, res})
- b.lock.Unlock()
-
- if res == nil {
- Logger.Printf("*** mockbroker/%d/%d: ignored %v", b.brokerID, idx, spew.Sdump(req))
- continue
- }
- Logger.Printf("*** mockbroker/%d/%d: served %v -> %v", b.brokerID, idx, req, res)
-
- encodedRes, err := encode(res, nil)
- if err != nil {
- b.serverError(err)
- break
- }
- if len(encodedRes) == 0 {
- b.lock.Lock()
- if b.notifier != nil {
- b.notifier(bytesRead, 0)
+ req, br, err := decodeRequest(bytes.NewReader(buffer))
+ bytesRead = br
+ if err != nil {
+ Logger.Printf("*** mockbroker/%d/%d: invalid request: err=%+v, %+v", b.brokerID, idx, err, spew.Sdump(req))
+ b.serverError(err)
+ break
}
- b.lock.Unlock()
- continue
- }
- binary.BigEndian.PutUint32(resHeader, uint32(len(encodedRes)+4))
- binary.BigEndian.PutUint32(resHeader[4:], uint32(req.correlationID))
- if _, err = conn.Write(resHeader); err != nil {
- b.serverError(err)
- break
- }
- if _, err = conn.Write(encodedRes); err != nil {
- b.serverError(err)
- break
+ if b.latency > 0 {
+ time.Sleep(b.latency)
+ }
+
+ b.lock.Lock()
+ res := b.handler(req)
+ b.history = append(b.history, RequestResponse{req.body, res})
+ b.lock.Unlock()
+
+ if res == nil {
+ Logger.Printf("*** mockbroker/%d/%d: ignored %v", b.brokerID, idx, spew.Sdump(req))
+ continue
+ }
+ Logger.Printf("*** mockbroker/%d/%d: served %v -> %v", b.brokerID, idx, req, res)
+
+ encodedRes, err := encode(res, nil)
+ if err != nil {
+ b.serverError(err)
+ break
+ }
+ if len(encodedRes) == 0 {
+ b.lock.Lock()
+ if b.notifier != nil {
+ b.notifier(bytesRead, 0)
+ }
+ b.lock.Unlock()
+ continue
+ }
+
+ binary.BigEndian.PutUint32(resHeader, uint32(len(encodedRes)+4))
+ binary.BigEndian.PutUint32(resHeader[4:], uint32(req.correlationID))
+ if _, err = conn.Write(resHeader); err != nil {
+ b.serverError(err)
+ break
+ }
+ if _, err = conn.Write(encodedRes); err != nil {
+ b.serverError(err)
+ break
+ }
+ bytesWritten = len(resHeader) + len(encodedRes)
+
+ } else {
+ // GSSAPI is not part of kafka protocol, but is supported for authentication proposes.
+ // Don't support history for this kind of request as is only used for test GSSAPI authentication mechanism
+ b.lock.Lock()
+ res := b.gssApiHandler(buffer)
+ b.lock.Unlock()
+ if res == nil {
+ Logger.Printf("*** mockbroker/%d/%d: ignored %v", b.brokerID, idx, spew.Sdump(buffer))
+ continue
+ }
+ if _, err = conn.Write(res); err != nil {
+ b.serverError(err)
+ break
+ }
+ bytesWritten = len(res)
}
b.lock.Lock()
if b.notifier != nil {
- b.notifier(bytesRead, len(resHeader)+len(encodedRes))
+ b.notifier(bytesRead, bytesWritten)
}
b.lock.Unlock()
+
}
Logger.Printf("*** mockbroker/%d/%d: connection closed, err=%v", b.BrokerID(), idx, err)
}