func (s *ScaleIOScheduler) subscribe() error {
	for {
		call := &sched.Call{
			Type: sched.Call_SUBSCRIBE.Enum(),
			Subscribe: &sched.Call_Subscribe{
				FrameworkInfo: s.Framework,
			},
		}
		resp, err := s.send(call)
		if err != nil {
			log.Warnln("subscribe error:", err)
		} else if resp != nil {
			go s.qEvents(resp)
			break
		}
		time.Sleep(time.Duration(subscribeRetryDelayInSec) * time.Second)
	}
	return nil
}
Subscribe subscribes the scheduler to the Mesos cluster.
It keeps the http connection opens with the Master to stream
subsequent events.
package scheduler
import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"time"
	log "github.com/Sirupsen/logrus"
	jsonpb "github.com/gogo/protobuf/jsonpb"
	"github.com/gogo/protobuf/proto"
	"github.com/codedellemc/scaleio-framework/scaleio-scheduler/client"
	"github.com/codedellemc/scaleio-framework/scaleio-scheduler/config"
	sched "github.com/codedellemc/scaleio-framework/scaleio-scheduler/mesos/sched"
	mesos "github.com/codedellemc/scaleio-framework/scaleio-scheduler/mesos/v1"
	kvstore "github.com/codedellemc/scaleio-framework/scaleio-scheduler/scheduler/kvstore"
	"github.com/codedellemc/scaleio-framework/scaleio-scheduler/scheduler/server"
)
const (
	subscribeRetryDelayInSec = 2
	rootKey                  = "scaleio-framework"
)
type ScaleIOScheduler struct {
	Config *config.Config
	Store  *kvstore.KvStore
	Framework *mesos.FrameworkInfo
	Server *server.RestServer
	Client *client.Client
	Events   chan *sched.Event
	DoneChan chan struct{}
}
func NewScaleIOScheduler(cfg *config.Config) *ScaleIOScheduler {
	myStore, err := kvstore.NewKvStore(cfg)
	if err != nil {
		log.Fatalln("NewKvStore Failed. Err:", err)
		return nil
	}
	if cfg.DeleteKeyValues {
		myStore.DeleteStore()
		return nil
	} else if cfg.DumpKeyValues {
		myStore.DumpStore()
		return nil
	} else if len(cfg.StoreAddKey) > 0 {
		err := myStore.UserKeyValue(cfg.StoreAddKey, cfg.StoreAddVal)
		if err == nil {
			log.Infoln("UserKeyValue Succeeded")
		} else {
			log.Errorln("UserKeyValue Failed. Err:", err)
		}
		return nil
	} else if len(cfg.StoreDelKey) > 0 {
		err := myStore.UserDeleteKey(cfg.StoreDelKey)
		if err == nil {
			log.Infoln("UserDeleteKey Succeeded")
		} else {
			log.Errorln("UserDeleteKey Failed. Err:", err)
		}
		return nil
	}
	return &ScaleIOScheduler{
		Config:    cfg,
		Store:     myStore,
		Client:    client.New(cfg.MasterREST, "/api/v1/scheduler"),
		Server:    server.NewRestServer(cfg, myStore),
		Framework: prepareFrameworkInfo(cfg),
		Events:    make(chan *sched.Event),
		DoneChan:  make(chan struct{}),
	}
}
func (s *ScaleIOScheduler) Start() <-chan struct{} {
	if err := s.subscribe(); err != nil {
		log.Errorln("Failed to subscribe:", err)
	}
	go s.handleEvents()
	return s.DoneChan
}
func (s *ScaleIOScheduler) Stop() {
	close(s.Events)
	close(s.DoneChan)
}
func (s *ScaleIOScheduler) send(call *sched.Call) (*http.Response, error) {
	marshaler := jsonpb.Marshaler{
		EnumsAsInts:  true,
		EmitDefaults: false,
		Indent:       "  ",
		OrigName:     false,
	}
	strJSON, errJSON := marshaler.MarshalToString(call)
	if errJSON == nil {
		log.Debugln("JSON:\n", strJSON)
	} else {
		log.Debugln("Unable to marshal to JSON:", errJSON)
	}
	payload, err := proto.Marshal(call)
	if err != nil {
		log.Errorln("Failed to Marshal Protobuf:", err)
		return nil, err
	}
	resp, err := s.Client.Send(payload)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
		msg := fmt.Sprint("StatusCode is not equal to StatusOK:", resp.StatusCode)
		log.Errorln(msg)
		return nil, errors.New(msg)
	}
	log.Infoln("StatusCode: StatusOK")
	return resp, nil
}