func (s *ZStoreStub) handleCreate(req *zc.ZMsg, resp *zc.ZMsg) {
	payload, _ := req.GetPayload()
	if !req.CheckString("zc-class") ||
		!req.CheckObject("zc-object") {
		resp.SetErr("invalid create request")
		log.Infof("invalid create request: %s", string(payload))
		return
	}

	className := req.GetString("zc-class")
	if !s.checkClass(className) {
		resp.SetErr("invalid class name")
		log.Infof("invalid create request: %s", string(payload))
		return
	}

	classStore := s.store[className]
	if classStore == nil {
		s.store[className] = make(map[string]zc.ZObject, 10)
		classStore = s.store[className]
	}

	zo := req.GetObject("zc-object")
	primaryKey := s.getPrimaryKey(className, zo)
	classStore[primaryKey] = zo
	resp.SetAck()
	resp.Put("zc-object", zo)
}

func (s *ZStoreStub) handleUpdate(req *zc.ZMsg, resp *zc.ZMsg) {
	payload, _ := req.GetPayload()
	if !req.CheckExists("zc-class", "zc-object") {
		resp.SetErr("invalid update request")
		log.Infof("invalid delete request: %s", string(payload))
		return
	}

	className := req.GetString("zc-class")
	if !s.checkClass(className) {
		resp.SetErr("invalid class name")
		log.Infof("invalid update request: %s", string(payload))
		return
	}

	classStore := s.store[className]
	if classStore == nil {
		s.store[className] = make(map[string]zc.ZObject, 10)
		classStore = s.store[className]
	}

	zo := req.GetObject("zc-object")
	primaryKey := s.getPrimaryKey(className, zo)
	oldZO := classStore[primaryKey]
	if oldZO != nil {
		for _, key := range zo.GetKeys() {
			oldZO.Put(key, zo.Get(key))
		}
	}
	resp.SetAck()
	return
}

func (s *ZStoreStub) handleDelete(req *zc.ZMsg, resp *zc.ZMsg) {
	payload, _ := req.GetPayload()
	if !req.CheckExists("zc-class", "zc-object") {
		resp.SetErr("invalid delete request")
		log.Infof("invalid delete request: %s", string(payload))
		return
	}

	className := req.GetString("zc-class")
	if !s.checkClass(className) {
		resp.SetErr("invalid class name")
		log.Infof("invalid create request: %s", string(payload))
		return
	}

	classStore := s.store[className]
	if classStore == nil || len(classStore) <= 0 {
		resp.SetAck()
		return
	}

	zo := req.GetObject("zc-object")
	primaryKey := s.getPrimaryKey(className, zo)
	delete(classStore, primaryKey)
	resp.SetAck()
	return
}

func getMap() map[string]int {
	var test map[string]int = make(map[string]int)
	test2 := test
	log.Infof("%p", test)
	log.Infof("%p", test2)
	return test2
}

// async send request to device and wait the response on result
func (this *Connection) SendRequest(packet *common.Message, result *chan common.Message) (err error) {
	common.Assert(packet != nil && result != nil, "check input param failed")
	oldId := packet.Header.MsgId
	log.Infof("receive new Request from app:msgId[%d]", oldId)
	request := NewDeviceCtrlRequest(oldId, result)
	// rewrite the msg id as the connection new id
	packet.Header.MsgId = this.getNewRequestId()
	err = this.requestMap.Insert(packet.Header.MsgId, request)
	if err != nil {
		log.Warningf("check request insert failed:old[%d], new[%d]", oldId, packet.Header.MsgId)
		return err
	}
	log.Infof("insert the request mapping succ:mid[%d], code[%d], dest[%v], gid[%s]",
		packet.Header.MsgId, packet.Header.MsgCode, this.socket.RemoteAddr(), this.gid.String())
	// if the dev connection closed panic will occured
	defer func() {
		packet.Header.MsgId = oldId
		info := recover()
		if info != nil {
			log.Warningf("the request queue is closed:err[%v]", info)
			err = common.ErrDeviceConnClosed
		}
	}()
	this.requestQueue <- *packet
	return nil
}

func (s *ZService) Start() error {
	log.Infof("begin to listen and servie: port=%s", s.port)
	err := http.ListenAndServe(":"+s.port, s)
	if err != nil {
		log.Infof("listenAndServe failed: %s", err.Error())
	}
	return err
}

func (c *ZServiceClient) Send(req *ZMsg) (resp *ZMsg, err error) {
	if req.hasObjectData() {
		err = req.encodeObject()
		if err != nil {
			return nil, err
		}
	}

	url := "http://" + c.serviceAddr + "/" + c.serviceName + "/" + req.GetName()
	payload, payloadFormat := req.GetPayload()
	log.Infof("client send req: serive=%s; name=%s; payload=%s",
		c.serviceName, req.GetName(), string(payload))

	httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(payload))
	if err != nil {
		return nil, errors.New("failed to create http req")
	}
	httpReq.Header.Set("Content-Type", payloadFormat)

	httpResp, err := http.DefaultClient.Do(httpReq)
	if err != nil {
		return nil, err
	}

	resp = NewZMsg()
	resp.SetName(httpResp.Header.Get(ZC_MSG_NAME_HEADER))

	if httpResp.ContentLength <= 0 {
		return resp, nil
	}
	if httpResp.Header.Get("Content-Type") != ZC_MSG_PAYLOAD_OBJECT {
		return nil, errors.New("wrong resp format, only support text/json")
	}

	respPayload := make([]byte, httpResp.ContentLength)
	l, err := httpResp.Body.Read(respPayload)
	if err != nil {
		return nil, err
	}
	if l != len(respPayload) {
		return nil, errors.New("failed to read http resp body")
	}

	log.Infof("client get resp: serive=%s; name=%s; payload=%s",
		c.serviceName, resp.GetName(), string(respPayload))
	resp.SetPayload(respPayload, httpResp.Header.Get("Content-Type"))
	err = resp.decodeObject()
	if err != nil {
		return nil, err
	}
	return resp, err
}

func (this *Connection) Loop(waitGroup *sync.WaitGroup) {
	// AFTER THIS ALL THE MESSAGE USING AES
	this.contex.EncryptType = common.ZC_SEC_TYPE_AES
	this.contex.SessionKey = this.sessionKey
	// process write queue in the new routine
	waitGroup.Add(1)
	go func() {
		// request packet
		var requestPacket common.Message
		for !this.exit {
			select {
			case requestPacket = <-this.requestQueue:
				// write the packet to read queue
				fmt.Println("[DEV] send the new Request Message to device:", requestPacket.Header.MsgId)
				err := common.Send(this.socket, this.contex, &requestPacket, common.DEV_WRITE_TIMEOUT)
				if err != nil {
					log.Warningf("forward the packet failed:err[%v]", err)
					this.exit = true
					break
				}
				// wait the empty response for the requestPacket
				// if not wait empty succ, can not send the next request
				// from the requestQueue....
				this.waitEmptyResponse(&requestPacket)
				fmt.Println("[DEV] wait Empty Response succ:", requestPacket.Header.MsgId)
			}
		}
		log.Infof("forward device request routine exit:dest[%v]", this.socket.RemoteAddr())
		waitGroup.Done()
	}()

	// process the read empty or response or report message
	var devMessage common.Message
	for !this.exit {
		// read the packet and dispatch to the processor
		err := common.Receive(this.socket, this.contex, &devMessage, common.DEV_READ_TIMEOUT)
		if err != nil {
			log.Warningf("read the packet failed:dest[%v], err[%v]", this.socket.RemoteAddr(), err)
			this.exit = true
			break
		}
		fmt.Println("[DEV] Receive message from device succ", devMessage.Header.String())
		// TODO if too long not receive dev message, close it
		// Handle Dev Message three different type
		// repsonse Ack + response Empty + request Report
		// WARNING:can not blocked here...
		this.handleDevMessage(&devMessage)
		fmt.Println("[DEV] Handle dev message finish:", time.Now().UTC(), devMessage.Header, this.socket.RemoteAddr())
	}
	log.Infof("device routine exit succ:dest[%v], gid[%s]", this.socket.RemoteAddr(), this.gid.String())
}

func (s *ZStoreStub) handleQuery(req *zc.ZMsg, resp *zc.ZMsg) {
	payload, _ := req.GetPayload()
	if !req.CheckExists("zc-class", "zc-query") {
		resp.SetErr("invalid query request")
		log.Infof("invalid query request: %s", string(payload))
		return
	}

	className := req.GetString("zc-class")
	if !s.checkClass(className) {
		resp.SetErr("invalid class name")
		log.Infof("invalid query request: %s", string(payload))
		return
	}

	classStore := s.store[className]
	if classStore == nil {
		resp.SetAck()
		return
	}

	query := req.GetObject("zc-query")
	if !query.CheckExists("zc-eq") {
		resp.SetErr("no eq condition to find objects")
		return
	}

	eq := query.GetObject("zc-eq")
	selectKeys := query.GetStrings("zc-select")
	primaryKey := s.getPrimaryKey(className, eq)
	dataZO := classStore[primaryKey]
	if dataZO == nil {
		resp.SetAck()
		return
	}

	selectZO := zc.NewObject()
	if len(selectKeys) <= 0 {
		selectZO = dataZO
	} else {
		for _, key := range selectKeys {
			selectZO.Put(key, dataZO.Get(key))
		}
	}
	resp.SetAck()
	resp.AddObject("zc-objects", selectZO)
	return
}

// if not find in database return nil + nil
func (this *WarehouseProxy) GetDeviceInfo(domain, subDomain, deviceId string) (*BasicInfo, error) {
	if this.cacheOn {
		basic, find := this.cache.Get(domain, subDomain, deviceId)
		if find {
			log.Infof("get device basic info from cache:domain[%s], device[%s:%s]", domain, subDomain, deviceId)
			return basic, nil
		}
	}
	SQL := fmt.Sprintf("SELECT device_type, public_key, status FROM %s_device_warehouse WHERE sub_domain = ? AND device_id = ?", domain)
	stmt, err := this.store.db.Prepare(SQL)
	if err != nil {
		log.Errorf("prepare query failed:err[%v]", err)
		return nil, err
	}
	defer stmt.Close()
	basic := NewBasicInfo()
	err = stmt.QueryRow(subDomain, deviceId).Scan(&basic.deviceType, &basic.publicKey, &basic.status)
	if err != nil {
		if err == sql.ErrNoRows {
			log.Warningf("no find the device:domain[%s], device[%s:%s]", domain, subDomain, deviceId)
			return nil, nil
		}
		log.Errorf("query failed:domain[%s], device[%s:%s]", domain, subDomain, deviceId)
		return nil, err
	}
	basic.subDomain = subDomain
	basic.deviceId = deviceId
	if this.cacheOn {
		this.cache.Set(domain, basic)
	}
	return basic, nil
}

// handle device response message forward to the result channel
func (this *Connection) handleDevResponse(packet *common.Message) {
	request, find := this.requestMap.Find(packet.Header.MsgId)
	if find {
		// reset to the old message id
		packet.Header.MsgId = request.(*DeviceCtrlRequest).oldMessageId
		defer func() {
			info := recover()
			if info != nil {
				log.Warningf("the response queue is closed:err[%v]", info)
			}
			_, exist := this.requestMap.Delete(packet.Header.MsgId)
			if !exist {
				log.Errorf("delete request failed:mid[%d], code[%d], dest[%s], gid[%s]",
					packet.Header.MsgId, packet.Header.MsgCode, this.socket.RemoteAddr(), this.gid.String())
			}
			log.Infof("delete the request mapping succ:mid[%d], code[%d], dest[%s], gid[%s]",
				packet.Header.MsgId, packet.Header.MsgCode, this.socket.RemoteAddr(), this.gid.String())
		}()
		// if closed will panic
		*(request.(*DeviceCtrlRequest).responseQueue) <- *packet
	} else {
		log.Errorf("check the dev response not find request:mid[%d], code[%d], dest[%s], gid[%s]",
			packet.Header.MsgId, packet.Header.MsgCode, this.socket.RemoteAddr(), this.gid.String())
	}
}

// the device point gateway start listen
func (this *DeviceGatewayServer) Start(host string) error {
	common.Assert((this.connManager != nil) && (this.devManager != nil), "check param nil")
	addr, err := net.ResolveTCPAddr("tcp4", host)
	if err != nil {
		log.Errorf("resolve tcp addr error:err[%v]", err)
		return err
	}
	ln, err := net.ListenTCP("tcp", addr)
	if err != nil {
		log.Errorf("listen error:err[%v]", err)
		return err
	}
	var waitGroup sync.WaitGroup
	for {
		socket, err := ln.AcceptTCP()
		if err != nil {
			log.Errorf("accept error:err[%v]", err)
			continue
		}
		log.Infof("device connect start:addr[%s]", socket.RemoteAddr())
		waitGroup.Add(1)
		go this.deviceRoutine(&waitGroup, socket, QueueLen)
	}
	waitGroup.Wait()
	return err
}

func (s *ZService) writeResp(resp *ZMsg, w http.ResponseWriter) {
	if resp.hasObjectData() {
		err := resp.encodeObject()
		if err != nil {
			w.Header().Set("Content-Length", "0")
			w.Header().Set(ZC_MSG_NAME_HEADER, ZC_MSG_NAME_ERR)
			return
		}
	}

	payload, payloadFormat := resp.GetPayload()
	if payload == nil {
		w.Header().Set("Content-Length", "0")
	} else {
		w.Header().Set("Content-Length", strconv.Itoa(len(payload)))
	}

	w.Header().Set("Content-Type", payloadFormat)
	w.Header().Set(ZC_MSG_NAME_HEADER, resp.GetName())

	if len(payload) > 0 {
		n, err := w.Write(payload)
		if n != len(payload) || err != nil {
			log.Infof("write payload failed: n=%d, err=%s", n, err.Error())
		}
	}
}

func (s *ZStoreStub) handleFind(req *zc.ZMsg, resp *zc.ZMsg) {
	payload, _ := req.GetPayload()
	if !req.CheckExists("zc-class", "zc-find") {
		resp.SetErr("invalid find request")
		log.Infof("invalid find request: %s", string(payload))
		return
	}

	className := req.GetString("zc-class")
	if !s.checkClass(className) {
		resp.SetErr("invalid class name")
		log.Infof("invalid find request: %s", string(payload))
		return
	}

	classStore := s.store[className]
	if classStore == nil {
		resp.SetAck()
		return
	}

	find := req.GetObject("zc-find")
	zo := find.GetObject("zc-object")
	selectKeys := find.GetStrings("zc-select")
	primaryKey := s.getPrimaryKey(className, zo)
	dataZO := classStore[primaryKey]
	if dataZO == nil {
		resp.SetAck()
		return
	}

	if len(selectKeys) <= 0 {
		resp.SetAck()
		resp.Put("zc-object", dataZO)
		return
	}

	selectZO := zc.NewObject()
	for _, key := range selectKeys {
		selectZO.Put(key, dataZO.Get(key))
	}
	resp.SetAck()
	resp.Put("zc-object", selectZO)
	return
}

//////////////////////////////////////////////////////////////////////////////
/// private interface related to database
//////////////////////////////////////////////////////////////////////////////
// transaction rollback according to the error status
func rollback(err *error, tx *sql.Tx) {
	if *err != nil {
		log.Infof("error occured rollback:err[%v]", *err)
		newErr := tx.Rollback()
		if newErr != nil {
			log.Errorf("rollback failed:err[%v]", newErr)
		}
	}
}

func TestZObject(t *testing.T) {
	o := NewZObject()
	o.PutString("name", "lihailei")
	o.PutInt("age", 1234)
	o.PutBool("vip", false)
	o.AddInt("myson", 1)
	o.AddInt("myson", 2)

	if o.GetString("name") != "lihailei" {
		t.Error("wrong name")
	}

	if o.GetBool("vip") != false {
		t.Error("wrong vip flag")
	}

	if len(o.GetBools("vips")) != 0 {
		t.Error("wrong vip flags")
	}

	if len(o.GetStrings("names")) != 0 {
		t.Error("wrong names")
	}

	o2 := NewZObject()
	o2.PutString("name", "lihailei2")
	o2.PutInt("age", 233)
	o2.PutBool("vip", true)
	o.AddObject("sons", o2)

	o3 := NewZObject()
	o3.PutString("name", "lihailei3")
	o3.PutInt("age", 23)
	o3.PutBool("vip", false)
	o.AddObject("sons", o3)

	opayload, _ := json.Marshal(o)
	o2payload, _ := json.Marshal(o2)
	o3payload, _ := json.Marshal(o3)

	log.Info("o: ", string(opayload))
	log.Info("o2: ", string(o2payload))
	log.Info("o3: ", string(o3payload))

	o4 := NewZObject()
	o5 := o4
	o4["testkey"] = "testvalue"
	log.Info(o4, o5)

	var gt = getMap()
	log.Infof("%p", gt)
	log.Info(len(gt))
	gt["lihailei"] = 2
	log.Info(gt)
}

func (this *GatewayServiceHandler) handleForwardRequest(req *zc.ZMsg, resp *zc.ZMsg) {
	domain := req.GetString("domain")
	subDomain := req.GetString("submain")
	deviceId := req.GetString("deviceid")
	Conn, find := this.device.Find(*server.NewDeviceGID(domain, subDomain, deviceId))
	if !find || Conn == nil {
		resp.SetErr(common.ErrDeviceForward.Error())
		log.Warningf("the device gateway not find:gid[%s:%s:%s]", domain, subDomain, deviceId)
		return
	}
	// convert the request to message by loading by so
	var request common.Message
	request.Header.Version = 1
	request.Header.MsgId = 1
	request.Header.MsgCode = 123
	payload, _ := req.GetPayload()
	request.Header.PayloadLen = uint16(len(payload))
	request.Payload = payload

	// receive the device response result
	result := make(chan common.Message, 32)
	defer close(result)
	err := Conn.(*server.Connection).SendRequest(&request, &result)
	if err != nil {
		resp.SetErr(err.Error())
		log.Warningf("forward message to dev failed:gid[%s:%s:%s], err[%v]", domain, subDomain, deviceId, err)
		return
	}
	log.Infof("forward message to device succ:code[%d], gid[%s:%s:%s]", request.Header.MsgCode, domain, subDomain, deviceId)

	// wait reponse from the device then response to the appclient
	response, err := this.waitResponse(request.Header.MsgId, result, common.APP_TIMEOUT)
	if err != nil {
		resp.SetErr(err.Error())
		log.Warningf("wait the current response failed:gid[%s:%s:%s], err[%v]", domain, subDomain, deviceId, err)
		return
	}
	log.Infof("forward device request succ:gid[%s:%s:%s]", domain, subDomain, deviceId)
	resp.PutObject("response", zc.ZObject{"body": response.Payload})
	resp.SetAck()
}

// delete home
func (this *HomeManagerHandler) handleDeleteHome(req *zc.ZMsg, resp *zc.ZMsg) {
	domain := req.GetString("domain")
	hid := req.GetInt("hid")
	err := this.home.Delete(domain, hid)
	if err != nil {
		resp.SetErr(err.Error())
		log.Warningf("delete home failed:domain[%s], hid[%d], err[%v]", domain, hid, err)
		return
	}
	log.Infof("delete home succ:domain[%s], hid[%d]", domain, hid)
	resp.SetAck()
}

func (this *DeviceGatewayServer) deviceRoutine(waitGroup *sync.WaitGroup, socket *net.TCPConn, maxLen int) {
	defer waitGroup.Done()
	conn := NewConnection(socket, maxLen, this.devManager)
	defer conn.Close()
	// step 1. shake hands with the dev connection
	deviceGid, err := conn.DeviceHandShake()
	if err != nil {
		if deviceGid != nil {
			log.Errorf("device handshake failed exit:addr[%s], gid[%s]", socket.RemoteAddr(), deviceGid.String())
			return
		}
		log.Errorf("device handshake failed exit:addr[%s]", socket.RemoteAddr())
		return
	}

	// step 2. set the socket option as keep alive
	socket.SetKeepAlive(true)
	socket.SetKeepAlivePeriod(time.Second * 30)

	// step 3. record the connection for forward manager
	err = this.connManager.Insert(deviceGid, conn)
	if err != nil {
		log.Errorf("insert the device connection Failed:addr[%s], gid[%s], err[%v]",
			socket.RemoteAddr(), deviceGid.String(), err)
		return
	}
	// for debug info
	log.Infof("device connection created:addr[%s], gid[%s]", socket.RemoteAddr(), deviceGid.String())

	// step 4. loop forward all request and receive all response
	conn.Loop(waitGroup)

	// step 5. remove conn from device connection manager
	_, find := this.connManager.Delete(deviceGid)
	if !find {
		log.Errorf("delete device connection before close:addr[%s], gid[%s]", socket.RemoteAddr(), deviceGid.String())
	}
	// for debug info
	log.Infof("device connection closed:addr[%s], gid[%s]", socket.RemoteAddr(), deviceGid.String())
}

// modify device
func (this *DeviceManagerHandler) handleModifyDevice(req *zc.ZMsg, resp *zc.ZMsg) {
	domain := req.GetString("domain")
	did := req.GetInt("did")
	name := req.GetString("dname")
	err := this.device.ChangeDeviceName(domain, did, name)
	if err != nil {
		resp.SetErr(err.Error())
		log.Warningf("modify device name failed:domain[%s], did[%d], name[%s], err[%v]", domain, did, name, err)
		return
	}
	log.Infof("modify device name succ:domain[%s], did[%s], name[%s]", domain, did, name)
	resp.SetAck()
}