add send msg param

zhangmeng

2019-08-30 383557064fd77c028fd71facf8ec1cf6b278d3e6

 nng.go

@@ -1,6 +1,7 @@
package deliver

import (
   "errors"
   "fmt"
   "os"
   "strings"
@@ -20,32 +21,28 @@
   "nanomsg.org/go-mangos/transport/all"
)

// type deliver
type td int

const (
   producer = td(iota)
   consumer
   star //mangos bus protocol
)

// NNG mangos wrap
type NNG struct {
   sock   mangos.Socket
   server bool
   mode   Mode
   url    string
   sock mangos.Socket
   typ  td
   mode Mode
   url  string

   sendMsg *mangos.Message

   arguments []interface{}
}

// Send impl interface Diliver
func (n *NNG) Send(data []byte) error {
   if n == nil {
      return errors.New("please init NNG first")
   }
   var err error
   if n.sock == nil {
      n.sock, err = n.makeNNG(producer)
      n.sock, err = n.makeNNG(agent)
      if err != nil {
         fmt.Println("create nng producer error")
         fmt.Println("create nng sender error")
         return err
      }
   }
@@ -54,20 +51,31 @@
      time.Sleep(time.Duration(surveyorTime*2) * time.Second)
   }

   msg := mangos.NewMessage(len(data))
   msg.Body = data
   return n.sock.SendMsg(msg)
   // msg := mangos.NewMessage(len(data))
   // msg.Body = data
   // return n.sock.SendMsg(msg)

   if n.sendMsg == nil {
      n.sendMsg = &mangos.Message{}
      n.sendMsg.Header = make([]byte, 32)
   }
   n.sendMsg.Body = data
   return n.sock.SendMsg(n.sendMsg)

}

// Recv impl interface Diliver
func (n *NNG) Recv() ([]byte, error) {
   if n == nil {
      return nil, errors.New("please init NNG first")
   }

   var err error

   if n.sock == nil {
      n.sock, err = n.makeNNG(consumer)
      n.sock, err = n.makeNNG(coactee)
      if err != nil {
         fmt.Println("create nng consumer error")
         fmt.Println("create nng reciever error")
         return nil, err
      }
   }
@@ -80,9 +88,16 @@

}

// Recv2 impl interface
func (n *NNG) Recv2(data []byte) (l int, err error) {
   data, err = n.Recv()
   l = len(data)
   return l, err
}

// Close impl interface Deliver
func (n *NNG) Close() {
   if n.sock != nil {
   if n != nil && n.sock != nil {
      n.sock.Close()
      n.sock = nil
   }
@@ -93,7 +108,7 @@
   rmExistedIpcName(url)

   return &NNG{
      server:    true,
      typ:       agent,
      mode:      m,
      url:       url,
      arguments: args,
@@ -103,7 +118,7 @@
func nngClient(m Mode, url string, args ...interface{}) *NNG {

   return &NNG{
      server:    false,
      typ:       coactee,
      mode:      m,
      url:       url,
      arguments: args,
@@ -112,98 +127,29 @@
}

func proto(typ td, m Mode) protocol {
   if typ == producer {
      return protoProducer(m)
   } else if typ == consumer {
      return protoConsumer(m)
   if typ == agent {
      return protoAgent(m)
   } else if typ == coactee {
      return protoCoactee(m)
   }
   return protoConsumer(m)
   return NONE
}

func (n *NNG) makeNNG(typ td) (mangos.Socket, error) {

   var sock mangos.Socket
   var err error
   if sock, err = newSocket(proto(typ, n.mode)); err != nil {
      return nil, err
   }

   if err = setSocketOptions(sock, n.arguments...); err != nil {
      sock.Close()
      sock = nil
   }
   if n.server {
      if err = sock.Listen(n.url); err != nil {
         sock.Close()
         sock = nil
      }
   } else {
      if err = sock.Dial(n.url); err != nil {
         sock.Close()
         sock = nil
      }
   }
   return sock, err
}

// maxRecvSize max recv size
var (
   maxRecvSize  = 33 * 1024 * 1024
   surveyorTime = -1
)

func defualtSocketOptions() map[string]interface{} {

   options := make(map[string]interface{})

   options[mangos.OptionMaxRecvSize] = maxRecvSize
   options[mangos.OptionWriteQLen] = 0
   options[mangos.OptionReadQLen] = 0
   options[mangos.OptionRecvDeadline] = time.Second
   options[mangos.OptionSendDeadline] = time.Second
   options[mangos.OptionRaw] = true

   return options
}

func setSocketOptions(sock mangos.Socket, args ...interface{}) error {

   options := defualtSocketOptions()

   switch sock.GetProtocol().Number() {
   case mangos.ProtoSub:
      topic := ""
      for _, arg := range args {
         switch arg.(type) {
         case string:
            topic = arg.(string)
         default:
         }
      }
      options[mangos.OptionSubscribe] = []byte(topic)
   case mangos.ProtoSurveyor:
      for _, arg := range args {
         switch arg.(type) {
         case int:
            if arg.(int) < 2 {
               surveyorTime = 1
            } else {
               surveyorTime = arg.(int) / 2
            }
         default:
         }
      }
      options[mangos.OptionSurveyTime] = time.Duration(surveyorTime) * time.Second

   switch n.mode {
   case Bus:
      sock, err = n.busMakeNNG(typ)
   case ReqRep, SurvResp:
      sock, err = n.rrMakeNNG(typ)
   default:
      sock, err = n.ppMakeNNG(typ)
   }

   for k, v := range options {
      if err := sock.SetOption(k, v); err != nil {
         return err
      }
   }
   return nil
   return sock, err
}

func rmExistedIpcName(url string) {
@@ -220,6 +166,9 @@
// access the underlying library.
func newSocket(p protocol) (mangos.Socket, error) {

   if p == NONE {
      return nil, errors.New("new socket protocol none")
   }
   var s mangos.Socket
   var err error

@@ -268,6 +217,7 @@

// Constants for protocols.
const (
   NONE       = -1
   PUSH       = protocol(mangos.ProtoPush)
   PULL       = protocol(mangos.ProtoPull)
   PUB        = protocol(mangos.ProtoPub)
@@ -280,38 +230,38 @@
   PAIR       = protocol(mangos.ProtoPair)
)

func protoProducer(m Mode) protocol {
func protoAgent(m Mode) protocol {
   switch m {
   case PushPull:
      return PUSH
   case PubSub:
      return PUB
   case ReqRep:
      return REP
   case SurvResp:
      return SURVEYOR
   case Bus:
      return BUS
   case Pair:
      return PAIR
   case SurvResp:
      return SURVEYOR
   case ReqRep:
      return REQ
   case Bus:
      return BUS
   }
   return PUSH
   return NONE
}

func protoConsumer(m Mode) protocol {
func protoCoactee(m Mode) protocol {
   switch m {
   case PushPull:
      return PULL
   case PubSub:
      return SUB
   case ReqRep:
      return REQ
   case SurvResp:
      return RESPONDENT
   case Bus:
      return BUS
   case Pair:
      return PAIR
   case SurvResp:
      return RESPONDENT
   case ReqRep:
      return REP
   case Bus:
      return BUS
   }
   return PULL
   return NONE
}