func (t *HTTPTransporter) Start() (upid.UPID, <-chan error) {
	t.stateLock.Lock()
	defer t.stateLock.Unlock()
	return t.state.Start()
}
Start starts the http transporter
package messenger
import (
	"bytes"
	"crypto/tls"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"
	log "github.com/golang/glog"
	"github.com/mesos/mesos-go/api/v0/upid"
	"golang.org/x/net/context"
)
const (
	DefaultReadTimeout  = 10 * time.Second
	DefaultWriteTimeout = 10 * time.Second
)
var (
	ReadTimeout  = DefaultReadTimeout
	WriteTimeout = DefaultWriteTimeout
	discardOnStopError = fmt.Errorf("discarding message because transport is shutting down")
	errNotStarted      = errors.New("HTTP transport has not been started")
	errTerminal        = errors.New("HTTP transport is terminated")
	errAlreadyRunning  = errors.New("HTTP transport is already running")
	httpTransport = http.Transport{
		Dial: (&net.Dialer{
			Timeout:   10 * time.Second,
			KeepAlive: 30 * time.Second,
		}).Dial,
		TLSHandshakeTimeout:   10 * time.Second,
		ResponseHeaderTimeout: DefaultReadTimeout,
	}
	HttpClient = http.Client{
		Timeout: DefaultReadTimeout,
	}
)
type httpTransporter interface {
	Send(ctx context.Context, msg *Message) error
	Recv() (*Message, error)
	Install(messageName string)
	Start() (upid.UPID, <-chan error)
	Stop(graceful bool) error
}
type notStartedState struct {
	h *HTTPTransporter
}
type stoppedState struct{}
type runningState struct {
	*notStartedState
}
func (s *notStartedState) Send(ctx context.Context, msg *Message) error { return errNotStarted }
func (s *notStartedState) Recv() (*Message, error)                      { return nil, errNotStarted }
func (s *notStartedState) Stop(graceful bool) error                     { return errNotStarted }
func (s *notStartedState) Install(messageName string)                   { s.h.install(messageName) }
func (s *notStartedState) Start() (upid.UPID, <-chan error) {
	s.h.state = &runningState{s}
	return s.h.start()
}
func (s *stoppedState) Send(ctx context.Context, msg *Message) error { return errTerminal }
func (s *stoppedState) Recv() (*Message, error)                      { return nil, errTerminal }
func (s *stoppedState) Stop(graceful bool) error                     { return errTerminal }
func (s *stoppedState) Install(messageName string)                   {}
func (s *stoppedState) Start() (upid.UPID, <-chan error) {
	ch := make(chan error, 1)
	ch <- errTerminal
	return upid.UPID{}, ch
}
func (s *runningState) Send(ctx context.Context, msg *Message) error { return s.h.send(ctx, msg) }
func (s *runningState) Recv() (*Message, error)                      { return s.h.recv() }
func (s *runningState) Stop(graceful bool) error {
	s.h.state = &stoppedState{}
	return s.h.stop(graceful)
}
func (s *runningState) Start() (upid.UPID, <-chan error) {
	ch := make(chan error, 1)
	ch <- errAlreadyRunning
	return upid.UPID{}, ch
}
type httpOpt func(*HTTPTransporter)
type HTTPTransporter struct {
	upid         upid.UPID
	listener     net.Listener  
	mux          *http.ServeMux
	tr           *http.Transport
	client       *http.Client
	messageQueue chan *Message
	address      net.IP  
	shouldQuit   chan struct{}
	stateLock    sync.RWMutex  
	state        httpTransporter
	server       *http.Server
}
func NewHTTPTransporter(upid upid.UPID, address net.IP, opts ...httpOpt) *HTTPTransporter {
	transport := httpTransport
	client := HttpClient
	client.Transport = &transport
	mux := http.NewServeMux()
	result := &HTTPTransporter{
		upid:         upid,
		messageQueue: make(chan *Message, defaultQueueSize),
		mux:          mux,
		client:       &client,
		tr:           &transport,
		address:      address,
		shouldQuit:   make(chan struct{}),
		server: &http.Server{
			ReadTimeout:  ReadTimeout,
			WriteTimeout: WriteTimeout,
			Handler:      mux,
		},
	}
	for _, f := range opts {
		f(result)
	}
	result.state = &notStartedState{result}
	return result
}
func ServerTLSConfig(config *tls.Config, nextProto map[string]func(*http.Server, *tls.Conn, http.Handler)) httpOpt {
	return func(transport *HTTPTransporter) {
		transport.server.TLSConfig = config
		transport.server.TLSNextProto = nextProto
	}
}
func ClientTLSConfig(config *tls.Config, handshakeTimeout time.Duration) httpOpt {
	return func(transport *HTTPTransporter) {
		transport.tr.TLSClientConfig = config
		transport.tr.TLSHandshakeTimeout = handshakeTimeout
	}
}
func (t *HTTPTransporter) getState() httpTransporter {
	t.stateLock.RLock()
	defer t.stateLock.RUnlock()
	return t.state
}
func (t *HTTPTransporter) Send(ctx context.Context, msg *Message) (sendError error) {
	return t.getState().Send(ctx, msg)
}
type mesosError struct {
	errorCode int
	upid      string
	uri       string
	status    string
}
func (e *mesosError) Error() string {
	return fmt.Sprintf("master %s rejected %s, returned status %q",
		e.upid, e.uri, e.status)
}
type networkError struct {
	cause error
}
func (e *networkError) Error() string {
	return e.cause.Error()
}
func (t *HTTPTransporter) send(ctx context.Context, msg *Message) (sendError error) {
	log.V(2).Infof("Sending message to %v via http\n", msg.UPID)
	req, err := t.makeLibprocessRequest(msg)
	if err != nil {
		return err
	}
	return t.httpDo(ctx, req, func(resp *http.Response, err error) error {
		if err != nil {
			log.V(1).Infof("Failed to POST: %v\n", err)
			return &networkError{err}
		}
		defer resp.Body.Close()
		if (resp.StatusCode != http.StatusOK) && (resp.StatusCode != http.StatusAccepted) {
			return &mesosError{
				errorCode: resp.StatusCode,
				upid:      msg.UPID.String(),
				uri:       msg.RequestURI(),
				status:    resp.Status,
			}
		}
		return nil
	})
}
func (t *HTTPTransporter) httpDo(ctx context.Context, req *http.Request, f func(*http.Response, error) error) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-t.shouldQuit:
		return discardOnStopError
	default:  
	}
	c := make(chan error, 1)
	go func() { c <- f(t.client.Do(req)) }()
	select {
	case <-ctx.Done():
		t.tr.CancelRequest(req)
		<-c  
		return ctx.Err()
	case err := <-c:
		return err
	case <-t.shouldQuit:
		t.tr.CancelRequest(req)
		<-c  
		return discardOnStopError
	}
}
func (t *HTTPTransporter) Recv() (*Message, error) {
	return t.getState().Recv()
}
func (t *HTTPTransporter) recv() (*Message, error) {
	select {
	default:
		select {
		case msg := <-t.messageQueue:
			return msg, nil
		case <-t.shouldQuit:
		}
	case <-t.shouldQuit:
	}
	return nil, discardOnStopError
}
func (t *HTTPTransporter) Install(msgName string) {
	t.getState().Install(msgName)
}
func (t *HTTPTransporter) install(msgName string) {
	requestURI := fmt.Sprintf("/%s/%s", t.upid.ID, msgName)
	t.mux.HandleFunc(requestURI, t.messageDecoder)
}
type loggedListener struct {
	delegate net.Listener
	done     <-chan struct{}
}
func (l *loggedListener) Accept() (c net.Conn, err error) {
	c, err = l.delegate.Accept()
	if c != nil {
		log.Infoln("accepted connection from", c.RemoteAddr())
		c = logConnection(c)
	} else if err != nil {
		select {
		case <-l.done:
		default:
			log.Errorln("failed to accept connection:", err.Error())
		}
	}
	return
}
func (l *loggedListener) Close() (err error) {
	err = l.delegate.Close()
	if err != nil {
		select {
		case <-l.done:
		default:
			log.Errorln("error closing listener:", err.Error())
		}
	} else {
		log.Infoln("closed listener")
	}
	return
}
func (l *loggedListener) Addr() net.Addr { return l.delegate.Addr() }
func logConnection(c net.Conn) net.Conn {
	w := hex.Dumper(os.Stdout)
	r := io.TeeReader(c, w)
	return &loggedConnection{
		Conn:   c,
		reader: r,
	}
}
type loggedConnection struct {
	net.Conn
	reader io.Reader
}
func (c *loggedConnection) Read(b []byte) (int, error) {
	return c.reader.Read(b)
}
func (t *HTTPTransporter) listen() error {
	var host string
	if t.address != nil {
		host = t.address.String()
	} else {
		host = t.upid.Host
	}
	var port string
	if t.upid.Port != "" {
		port = t.upid.Port
	} else {
		port = "0"
	}
	ln, err := net.Listen("tcp4", net.JoinHostPort(host, port))
	if err != nil {
		log.Errorf("HTTPTransporter failed to listen: %v\n", err)
		return err
	}
	host, port, _ = net.SplitHostPort(ln.Addr().String())
	log.Infoln("listening on", host, "port", port)
	if len(t.upid.Host) == 0 {
		t.upid.Host = host
	}
	if len(t.upid.Port) == 0 || t.upid.Port == "0" {
		t.upid.Port = port
	}
	if log.V(3) {
		t.listener = &loggedListener{delegate: ln, done: t.shouldQuit}
	} else {
		t.listener = ln
	}
	return nil
}