func (ic *IbClient) CancelMktData(reqID int64) {
	const v = 2
	fields := make([]interface{}, 0, 3)
	fields = append(fields,
		mCANCEL_MKT_DATA,
		v,
		reqID,
	)
	msg := makeMsgBytes(fields...)
	ic.reqChan <- msg
}
CancelMktData cancels the market data
package ibapi
import (
	"bufio"
	"bytes"
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"go.uber.org/zap"
)
const (
	MaxRequests = 95
	RequestInternal = 2
	MaxClientVersion = 148
)
type IbClient struct {
	host             string
	port             int
	clientID         int64
	conn             *IbConnection
	scanner          *bufio.Scanner
	writer           *bufio.Writer
	wrapper          IbWrapper
	decoder          ibDecoder
	connectOptions   string
	reqIDSeq         int64
	reqChan          chan []byte
	errChan          chan error
	msgChan          chan []byte
	timeChan         chan time.Time
	terminatedSignal chan int   
	done             chan bool  
	clientVersion    Version
	serverVersion    Version
	connTime         string
	extraAuth        bool
	wg               sync.WaitGroup
	ctx              context.Context
	err              error
}
func NewIbClient(wrapper IbWrapper) *IbClient {
	ic := &IbClient{}
	ic.SetWrapper(wrapper)
	ic.reset()
	return ic
}
func (ic *IbClient) ConnState() int {
	return ic.conn.state
}
func (ic *IbClient) setConnState(connState int) {
	preState := ic.conn.state
	ic.conn.state = connState
	log.Debug("change connection state", zap.Int("previous", preState), zap.Int("current", connState))
}
func (ic *IbClient) GetReqID() int64 {
	return atomic.AddInt64(&ic.reqIDSeq, 1)
}
func (ic *IbClient) SetWrapper(wrapper IbWrapper) {
	ic.wrapper = wrapper
	log.Debug("set wrapper", zap.Reflect("wrapper", wrapper))
	ic.decoder = ibDecoder{wrapper: ic.wrapper}
}
func (ic *IbClient) SetContext(ctx context.Context) {
	ic.ctx = ctx
}
func (ic *IbClient) SetConnectionOptions(opts string) {
	ic.connectOptions = opts
}
func (ic *IbClient) Connect(host string, port int, clientID int64) error {
	ic.host, ic.port, ic.clientID = host, port, clientID
	log.Info("Connect to client", zap.String("host", host), zap.Int("port", port), zap.Int64("clientID", clientID))
	ic.setConnState(CONNECTING)
	if err := ic.conn.connect(host, port); err != nil {
		ic.wrapper.Error(NO_VALID_ID, CONNECT_FAIL.code, CONNECT_FAIL.msg)
		ic.reset()
		return CONNECT_FAIL
	}
	ic.done = make(chan bool)
	return nil
}
func (ic *IbClient) Disconnect() error {
	log.Debug("close terminatedSignal chan")
	close(ic.terminatedSignal)  
	if err := ic.conn.disconnect(); err != nil {
		return err
	}
	ic.wg.Wait()
	defer func() {
		ic.done <- true
	}()
	defer ic.reset()
	defer ic.wrapper.ConnectionClosed()
	defer log.Info("Disconnected!")
	return ic.err
}
func (ic *IbClient) IsConnected() bool {
	return ic.conn.state == CONNECTED
}
func (ic *IbClient) startAPI() error {
	var startAPI []byte
	const v = 2
	if ic.serverVersion >= mMIN_SERVER_VER_OPTIONAL_CAPABILITIES {
		startAPI = makeMsgBytes(mSTART_API, v, ic.clientID, "")
	} else {
		startAPI = makeMsgBytes(mSTART_API, v, ic.clientID)
	}
	log.Debug("start API", zap.Binary("bytes", startAPI))
	if _, err := ic.writer.Write(startAPI); err != nil {
		return err
	}
	return ic.writer.Flush()
}
func (ic *IbClient) HandShake() error {
	log.Info("HandShake with TWS or GateWay")
	var msg bytes.Buffer
	var msgBytes []byte
	head := []byte("API\x00")
	connectOptions := ""
	if ic.connectOptions != "" {
		connectOptions = " " + ic.connectOptions
	}
	clientVersion := []byte(fmt.Sprintf("v%d..%d%s", MIN_CLIENT_VER, MAX_CLIENT_VER, connectOptions))
	sizeofCV := make([]byte, 4)
	binary.BigEndian.PutUint32(sizeofCV, uint32(len(clientVersion)))
	msg.Write(head)
	msg.Write(sizeofCV)
	msg.Write(clientVersion)
	log.Debug("send handShake header", zap.Binary("header", msg.Bytes()))
	if _, err := ic.writer.Write(msg.Bytes()); err != nil {
		return err
	}
	if err := ic.writer.Flush(); err != nil {
		return err
	}
	log.Debug("recv handShake Info")
	if !ic.scanner.Scan() {
		return ic.scanner.Err()
	}
	msgBytes = ic.scanner.Bytes()
	serverInfo := splitMsgBytes(msgBytes)
	v, _ := strconv.Atoi(string(serverInfo[0]))
	ic.serverVersion = Version(v)
	ic.connTime = string(serverInfo[1])
	ic.decoder.setVersion(ic.serverVersion)
	ic.decoder.setmsgID2process()
	log.Info("handShake info", zap.Int("serverVersion", ic.serverVersion))
	log.Info("handShake info", zap.String("connectionTime", ic.connTime))
	if err := ic.startAPI(); err != nil {
		return err
	}
	go ic.goReceive()  
	comfirmMsgIDs := []IN{mNEXT_VALID_ID, mMANAGED_ACCTS}
	timeout := time.After(60 * time.Second)
comfirmReadyLoop:
	for {
		select {
		case m := <-ic.msgChan:
			f := splitMsgBytes(m)
			MsgID, _ := strconv.ParseInt(string(f[0]), 10, 64)
			ic.decoder.interpret(m)
			for i, ID := range comfirmMsgIDs {
				if MsgID == ID {
					comfirmMsgIDs = append(comfirmMsgIDs[:i], comfirmMsgIDs[i+1:]...)
					break
				}
			}
			if len(comfirmMsgIDs) == 0 {
				ic.setConnState(CONNECTED)
				ic.wrapper.ConnectAck()
				break comfirmReadyLoop
			}
		case <-timeout:
			ic.setConnState(DISCONNECTED)
			ic.wrapper.Error(NO_VALID_ID, ALREADY_CONNECTED.code, ALREADY_CONNECTED.msg)
			return ALREADY_CONNECTED
		case <-ic.ctx.Done():
			ic.setConnState(DISCONNECTED)
			ic.wrapper.Error(NO_VALID_ID, ALREADY_CONNECTED.code, ALREADY_CONNECTED.msg)
			return ALREADY_CONNECTED
		}
	}
	log.Info("HandShake completed")
	return nil
}
func (ic *IbClient) ServerVersion() Version {
	return ic.serverVersion
}
func (ic *IbClient) ConnectionTime() string {
	return ic.connTime
}
func (ic *IbClient) reset() {
	log.Debug("reset ibClient")
	ic.reqIDSeq = 0
	ic.conn = &IbConnection{}
	ic.host = ""
	ic.port = -1
	ic.extraAuth = false
	ic.clientID = -1
	ic.serverVersion = -1
	ic.connTime = ""
	ic.scanner = bufio.NewScanner(ic.conn)
	ic.scanner.Split(scanFields)
	ic.scanner.Buffer(make([]byte, 4096), MAX_MSG_LEN)
	ic.writer = bufio.NewWriter(ic.conn)
	ic.reqChan = make(chan []byte, 10)
	ic.errChan = make(chan error, 10)
	ic.msgChan = make(chan []byte, 100)
	ic.terminatedSignal = make(chan int)
	ic.wg = sync.WaitGroup{}
	ic.connectOptions = ""
	ic.setConnState(DISCONNECTED)
	ic.err = nil
	if ic.ctx == nil {
		ic.ctx = context.TODO()
	}
}
func (ic *IbClient) SetServerLogLevel(logLevel int64) {
	const v = 1
	fields := make([]interface{}, 0, 3)
	fields = append(fields,
		mSET_SERVER_LOGLEVEL,
		v,
		logLevel,
	)
	msg := makeMsgBytes(fields...)
	ic.reqChan <- msg
}
func (ic *IbClient) ReqMktData(reqID int64, contract *Contract, genericTickList string, snapshot bool, regulatorySnapshot bool, mktDataOptions []TagValue) {
	switch {
	case ic.serverVersion < mMIN_SERVER_VER_DELTA_NEUTRAL && contract.DeltaNeutralContract != nil:
		ic.wrapper.Error(reqID, UPDATE_TWS.code, UPDATE_TWS.msg+"  It does not support delta-neutral orders.")
		return
	case ic.serverVersion < mMIN_SERVER_VER_REQ_MKT_DATA_CONID && contract.ContractID > 0:
		ic.wrapper.Error(reqID, UPDATE_TWS.code, UPDATE_TWS.msg+"  It does not support conId parameter.")
		return
	case ic.serverVersion < mMIN_SERVER_VER_TRADING_CLASS && contract.TradingClass != "":
		ic.wrapper.Error(reqID, UPDATE_TWS.code, UPDATE_TWS.msg+"  It does not support tradingClass parameter in reqMktData.")
		return
	}
	const v = 11
	fields := make([]interface{}, 0, 30)
	fields = append(fields,
		mREQ_MKT_DATA,
		v,
		reqID,
	)
	if ic.serverVersion >= mMIN_SERVER_VER_REQ_MKT_DATA_CONID {
		fields = append(fields, contract.ContractID)
	}
	fields = append(fields,
		contract.Symbol,
		contract.SecurityType,
		contract.Expiry,
		contract.Strike,
		contract.Right,
		contract.Multiplier,
		contract.Exchange,
		contract.PrimaryExchange,
		contract.Currency,
		contract.LocalSymbol)
	if ic.serverVersion >= mMIN_SERVER_VER_TRADING_CLASS {
		fields = append(fields, contract.TradingClass)
	}
	if contract.SecurityType == "BAG" {
		comboLegsCount := len(contract.ComboLegs)
		fields = append(fields, comboLegsCount)
		for _, comboLeg := range contract.ComboLegs {
			fields = append(fields,
				comboLeg.ContractID,
				comboLeg.Ratio,
				comboLeg.Action,
				comboLeg.Exchange)
		}
	}
	if ic.serverVersion >= mMIN_SERVER_VER_DELTA_NEUTRAL {
		if contract.DeltaNeutralContract != nil {
			fields = append(fields,
				true,
				contract.DeltaNeutralContract.ContractID,
				contract.DeltaNeutralContract.Delta,
				contract.DeltaNeutralContract.Price)
		} else {
			fields = append(fields, false)
		}
	}
	fields = append(fields,
		genericTickList,
		snapshot)
	if ic.serverVersion >= mMIN_SERVER_VER_REQ_SMART_COMPONENTS {
		fields = append(fields, regulatorySnapshot)
	}
	if ic.serverVersion >= mMIN_SERVER_VER_LINKING {
		if len(mktDataOptions) > 0 {
			log.Panic("not supported")
		}
		fields = append(fields, "")
	}
	msg := makeMsgBytes(fields...)
	ic.reqChan <- msg
}