func NewMockBroker(t TestState, brokerID int) *MockBroker {
	var err error
	broker := &MockBroker{
		stopper:      make(chan bool),
		t:            t,
		brokerID:     brokerID,
		expectations: make(chan encoder, 512),
	}
	broker.listener, err = net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatal(err)
	}
	_, portStr, err := net.SplitHostPort(broker.listener.Addr().String())
	if err != nil {
		t.Fatal(err)
	}
	tmp, err := strconv.ParseInt(portStr, 10, 32)
	if err != nil {
		t.Fatal(err)
	}
	broker.port = int32(tmp)
	go broker.serverLoop()
	return broker
}
NewMockBroker launches a fake Kafka broker. It takes a TestState (e.g. *testing.T) as provided by the
test framework and a channel of responses to use.  If an error occurs it is
simply logged to the TestState and the broker exits.
package sarama
import (
	"encoding/binary"
	"errors"
	"io"
	"net"
	"strconv"
)
type TestState interface {
	Error(args ...interface{})
	Fatal(args ...interface{})
	Fatalf(format string, args ...interface{})
}
type MockBroker struct {
	brokerID     int
	port         int32
	stopper      chan bool
	expectations chan encoder
	listener     net.Listener
	t            TestState
	expecting    encoder
}
func (b *MockBroker) BrokerID() int {
	return b.brokerID
}
func (b *MockBroker) Port() int32 {
	return b.port
}
func (b *MockBroker) Addr() string {
	return b.listener.Addr().String()
}
type rawExpectation []byte
func (r rawExpectation) ResponseBytes() []byte {
	return r
}
func (b *MockBroker) Close() {
	if b.expecting != nil {
		b.t.Fatalf("Not all expectations were satisfied in mockBroker with ID=%d! Still waiting on %#v", b.BrokerID(), b.expecting)
	}
	close(b.expectations)
	<-b.stopper
}
func (b *MockBroker) serverLoop() (ok bool) {
	var (
		err  error
		conn net.Conn
	)
	defer close(b.stopper)
	if conn, err = b.listener.Accept(); err != nil {
		return b.serverError(err, conn)
	}
	reqHeader := make([]byte, 4)
	resHeader := make([]byte, 8)
	for expectation := range b.expectations {
		b.expecting = expectation
		_, err = io.ReadFull(conn, reqHeader)
		b.expecting = nil
		if err != nil {
			return b.serverError(err, conn)
		}
		body := make([]byte, binary.BigEndian.Uint32(reqHeader))
		if len(body) < 10 {
			return b.serverError(errors.New("Kafka request too short."), conn)
		}
		if _, err = io.ReadFull(conn, body); err != nil {
			return b.serverError(err, conn)
		}
		response, err := encode(expectation)
		if err != nil {
			return false
		}
		if len(response) == 0 {
			continue
		}
		binary.BigEndian.PutUint32(resHeader, uint32(len(response)+4))
		binary.BigEndian.PutUint32(resHeader[4:], binary.BigEndian.Uint32(body[4:]))
		if _, err = conn.Write(resHeader); err != nil {
			return b.serverError(err, conn)
		}
		if _, err = conn.Write(response); err != nil {
			return b.serverError(err, conn)
		}
	}
	if err = conn.Close(); err != nil {
		return b.serverError(err, nil)
	}
	if err = b.listener.Close(); err != nil {
		b.t.Error(err)
		return false
	}
	return true
}
func (b *MockBroker) serverError(err error, conn net.Conn) bool {
	b.t.Error(err)
	if conn != nil {
		conn.Close()
	}
	b.listener.Close()
	return false
}