func (b *Broker) Open(conf *BrokerConfig) error {
	if conf == nil {
		conf = NewBrokerConfig()
	}
	err := conf.Validate()
	if err != nil {
		return err
	}
	b.lock.Lock()
	if b.conn != nil {
		b.lock.Unlock()
		Logger.Printf("Failed to connect to broker %s\n", b.addr)
		Logger.Println(AlreadyConnected)
		return AlreadyConnected
	}
	go withRecover(func() {
		defer b.lock.Unlock()
		b.conn, b.connErr = net.Dial("tcp", b.addr)
		if b.connErr != nil {
			Logger.Printf("Failed to connect to broker %s\n", b.addr)
			Logger.Println(b.connErr)
			return
		}
		b.conf = conf
		b.done = make(chan bool)
		b.responses = make(chan responsePromise, b.conf.MaxOpenRequests)
		Logger.Printf("Connected to broker %s\n", b.addr)
		go withRecover(b.responseReceiver)
	})
	return nil
}
Open tries to connect to the Broker. It takes the broker lock synchronously, then spawns a goroutine which
connects and releases the lock. This means any subsequent operations on the broker will block waiting for
the connection to finish. To get the effect of a fully synchronous Open call, follow it by a call to Connected().
The only errors Open will return directly are ConfigurationError or AlreadyConnected. If conf is nil, the result of
NewBrokerConfig() is used.
package sarama
import (
	"fmt"
	"io"
	"net"
	"strconv"
	"sync"
	"time"
)
type BrokerConfig struct {
	MaxOpenRequests int            
	ReadTimeout     time.Duration  
	WriteTimeout    time.Duration  
}
func NewBrokerConfig() *BrokerConfig {
	return &BrokerConfig{
		MaxOpenRequests: 4,
		ReadTimeout:     1 * time.Minute,
		WriteTimeout:    1 * time.Minute,
	}
}
func (config *BrokerConfig) Validate() error {
	if config.MaxOpenRequests < 0 {
		return ConfigurationError("Invalid MaxOpenRequests")
	}
	if config.ReadTimeout <= 0 {
		return ConfigurationError("Invalid ReadTimeout")
	}
	if config.WriteTimeout <= 0 {
		return ConfigurationError("Invalid WriteTimeout")
	}
	return nil
}
type Broker struct {
	id   int32
	addr string
	conf          *BrokerConfig
	correlationID int32
	conn          net.Conn
	connErr       error
	lock          sync.Mutex
	responses chan responsePromise
	done      chan bool
}
type responsePromise struct {
	correlationID int32
	packets       chan []byte
	errors        chan error
}
func NewBroker(addr string) *Broker {
	return &Broker{id: -1, addr: addr}
}