copy from bgr-2-analysis

zhangmeng

2020-01-15 d85f3edab0d8c495cecd7a81f31a9ead1eb001ac

 common/send.go

@@ -4,14 +4,14 @@
   "context"
   "time"

   "basic.com/libgowrapper/sdkstruct.git"
   "basic.com/valib/deliver.git"
   "github.com/gogo/protobuf/proto"
)

// Sender decoder ingo
type Sender struct {
   ipcURL string
   chMsg  <-chan sdkstruct.MsgSDK
   chMsg  <-chan MsgRS
   shm    bool
   fn     func([]byte, bool)

@@ -27,7 +27,7 @@
}

// NewSender Sender
func NewSender(ipcURL string, chMsg <-chan sdkstruct.MsgSDK, shm bool, fn func(...interface{})) *Sender {
func NewSender(ipcURL string, chMsg <-chan MsgRS, shm bool, fn func(...interface{})) *Sender {
   // logo.Infof("create ipc %s for decode : %s\n", ipcURL, ipcURL)
   return &Sender{
      ipcURL:   ipcURL,
@@ -62,19 +62,28 @@
         return
      case i := <-s.chMsg:

         data <- i.MsgData
         d, err := proto.Marshal(&i.Msg)

         if int(i.SdkIndex+1) == i.SdkCount {
         if err != nil {
            s.fnLogger("protobuf encode ipc sender error: ", err)
            continue
         }

         data <- d

         if int(i.Msg.Tasklab.Index+1) == len(i.Msg.Tasklab.Sdkinfos) {
            if s.fn != nil {

               sFlag := true
               if i.SdkDataLen < 2 {
               for _, v := range i.Msg.Tasklab.Sdkinfos {
                  if len(v.Sdkdata) < 2 {
                  sFlag = false
                     break
               }
               s.fn(i.MsgData, sFlag)
               }
               s.fn(d, sFlag)
            }
         }

            }
         }
      default:
         time.Sleep(10 * time.Millisecond)
      }
@@ -82,8 +91,6 @@
}

func (s *Sender) run(ctx context.Context, i deliver.Deliver) {

   // go ruleserver.TimeTicker()

   dataChan := make(chan []byte, 3)
   go s.serializeProto(ctx, dataChan)