add SaveNotExistOrUpdate

huchanghua

2022-07-22 293b6d664512c30d2e6b2ce80487ba6fa16598c3

 sbusClient.go

@@ -1,15 +1,15 @@
package bhomedbapi

import (
   "basic.com/valib/bhomebus.git"
   "basic.com/valib/c_bhomebus.git/api/bhsgo"
   "basic.com/valib/c_bhomebus.git/proto/source/bhome_msg"
   "encoding/json"
   "errors"
   "fmt"
   "strconv"
)

type SBusClient struct {
   nodes []bhomebus.NetNode
   nodes []*bhome_msg.MsgQueryTopicReply_BHNodeAddress
}

type ProcInfo struct {
@@ -18,14 +18,7 @@
   Info          string             `json:"info"` // 进程的描述信息，用于区分同一进程名称下多个进程
}

type MsgInfo struct {
   SrcProc        ProcInfo            `json:"srcProc"`       // 源进程基本信息
   MsgType         string             `json:"msgType"`        // 数据类型，可为请求、发布、订阅、应答等
   Topic          string            `json:"topic"`         //请求主题
   Body          []byte            `json:"body"`         //请求内容或者反馈结果
}

type request struct {
type Request struct {
   Path                 string                 `json:"path"`
   Method               string                 `json:"method"`
   ContentType          string                 `json:"contentType"`
@@ -36,26 +29,37 @@
   Body                 []byte                 `json:"body"`
}

func (sc SBusClient) DoGetRequest(url string, params map[string]string, headers map[string]string) ([]byte, error) {
   if sc.nodes == nil || len(sc.nodes) == 0 {
      return nil, errors.New("invalid netNodes")
   }
type reply struct {
   Success          bool                   `json:"success"`
   Msg             string                   `json:"msg"`
   Data             interface{}             `json:"data"`
}

   req := request {
func (sc SBusClient) DoGetRequest(url string, params map[string]string, headers map[string]string) ([]byte, error) {

   req := Request{
      Path:        url,
      Method:      "GET",
      ContentType: CONTENT_TYPE_JSON,
   }
   fillParam(&req, headers, params, nil)

   return doReq(req, sc.nodes)
   rb, err := json.Marshal(req)
   if err !=nil {
      return nil,err
   }
   rMsg := bhome_msg.MsgRequestTopic{
      Topic: []byte(req.Path),
      Data: rb,
   }

   return busReq(&rMsg, sc.nodes)
   //return doReq(req, sc.nodes)
}

func (sc SBusClient) DoPostRequest(url string, contentType string, body map[string]interface{}, params map[string]string, headers map[string]string) ([]byte, error) {
   if sc.nodes == nil || len(sc.nodes) == 0 {
      return nil, errors.New("invalid port")
   }
   req := request {

   req := Request{
      Path:        url,
      Method:      "POST",
      ContentType: contentType,
@@ -64,6 +68,7 @@
   if contentType == CONTENT_TYPE_FORM || contentType == CONTENT_TYPE_MULFORM {
      if body != nil {
         req.PostFormMap = make(map[string][]string)
         req.FormMap = map[string][]string{}
         for k,v := range body {
            switch v.(type) {
            case string:
@@ -90,38 +95,61 @@
      }
   }

   return doReq(req, sc.nodes)
   rb, err := json.Marshal(req)
   if err !=nil {
      return nil,err
   }
   rMsg := bhome_msg.MsgRequestTopic{
      Topic: []byte(req.Path),
      Data: rb,
   }

   return busReq(&rMsg, sc.nodes)
}

func (sc SBusClient) DoPutRequest(url string, contentType string, body map[string]interface{}, headers map[string]string) ([]byte, error) {
   if sc.nodes == nil || len(sc.nodes) == 0 {
      return nil, errors.New("invalid port")
   }
   req := request {

   req := Request{
      Path:        url,
      Method:      "PUT",
      ContentType: contentType,
   }
   fillParam(&req, headers, nil, body)

   return doReq(req, sc.nodes)
   rb, err := json.Marshal(req)
   if err !=nil {
      return nil,err
   }
   rMsg := bhome_msg.MsgRequestTopic{
      Topic: []byte(req.Path),
      Data: rb,
   }

   return busReq(&rMsg, sc.nodes)
}

func (sc SBusClient) DoDeleteRequest(url string, contentType string, body map[string]interface{}, headers map[string]string) ([]byte, error) {
   if sc.nodes == nil || len(sc.nodes) == 0 {
      return nil, errors.New("invalid port")
   }
   req := request {

   req := Request{
      Path:        url,
      Method:      "DELETE",
      ContentType: contentType,
   }
   fillParam(&req, headers, nil, body)

   return doReq(req, sc.nodes)
   rb, err := json.Marshal(req)
   if err !=nil {
      return nil,err
   }
   rMsg := bhome_msg.MsgRequestTopic{
      Topic: []byte(req.Path),
      Data: rb,
   }

   return busReq(&rMsg, sc.nodes)
}

func fillParam(req *request,headers map[string]string, params map[string]string, body map[string]interface{}) {
func fillParam(req *Request,headers map[string]string, params map[string]string, body map[string]interface{}) {
   headerMap := make(map[string][]string)
   queryMap := make(map[string][]string)
   if headers != nil {
@@ -142,38 +170,22 @@
   } else {
      req.Body = b
   }

}

func doReq(req request, nodes []bhomebus.NetNode) ([]byte,error) {
   rb, err := json.Marshal(req)
   if err !=nil {
      return nil,err
/**
* 新版bhs直接发起请求，不会产生共享内存泄漏的问题？
*/
func doReq(req *bhome_msg.MsgRequestTopic, destArr []bhome_msg.BHAddress) ([]byte, error) {
   dest := bhome_msg.BHAddress{}
   if destArr != nil && len(destArr) > 0 {
      dest = destArr[0]
   }
   rMsg := MsgInfo{
      Topic: req.Path,
      Body: rb,
   }
   data, err := json.Marshal(rMsg)
   if err != nil {
      return nil, err
   }
   s := bhomebus.OpenSocket()
   defer s.Close()
   var ret []bhomebus.Mesg
   if n := s.SendandrecvTimeout(nodes, data, &ret, 5000);n == 0 {  //n==0表示没有请求成功
      return nil, fmt.Errorf("doReq s.SendandrecvTimeout result n:%d", n)
   pid := ""
   r := bhome_msg.MsgRequestTopicReply{}
   if bhsgo.Request(&dest, req, &pid, &r, 5000) {
      return r.Data, nil
   } else {
      logPrint("doReq SendandrecvTimeout n:", n)
      if len(ret) > 0 {
         var retMsg MsgInfo
         err = json.Unmarshal(ret[0].Data, &retMsg)
         if err != nil {
            logPrint("doReq unmarshal to MsgInfo err:", err)
            return nil, err
         }
         return retMsg.Body, nil
      }
      return nil, fmt.Errorf("no any response")
      logPrint("bhsgo.Request request err:", r.Errmsg)
      return nil, errors.New("bhsgo.Request return false")
   }
}