From 9e1301abf98fb4e04fabba535c1bb5ad161a66a4 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期三, 15 五月 2019 12:24:02 +0800
Subject: [PATCH] add surveyor
---
nng.go | 38 ++++++++++++++++++++++++++++++--------
1 files changed, 30 insertions(+), 8 deletions(-)
diff --git a/nng.go b/nng.go
index e9568cd..01539a3 100644
--- a/nng.go
+++ b/nng.go
@@ -5,6 +5,7 @@
"fmt"
"os"
"strings"
+ "time"
"nanomsg.org/go-mangos"
"nanomsg.org/go-mangos/protocol/bus"
@@ -31,11 +32,17 @@
return errors.New("please init NNG first")
}
+ switch n.sock.GetProtocol().Number() {
+ case mangos.ProtoSurveyor:
+ time.Sleep(surveyorTime * 2)
+ default:
+ }
if _, err := n.sock.GetOption(mangos.OptionRaw); err == nil {
msg := mangos.NewMessage(len(data))
msg.Body = data
return n.sock.SendMsg(msg)
}
+
return n.sock.Send(data)
}
@@ -51,8 +58,8 @@
return n.sock.Recv()
}
-// NewNNGProducer create from deliver Mode
-func NewNNGProducer(m Mode, url string, args ...interface{}) *NNG {
+// nngProducer create from deliver Mode
+func nngProducer(m Mode, url string, args ...interface{}) *NNG {
rmExistedIpcName(url)
if sock, err := newSocket(protoProducer(m)); err == nil {
@@ -71,8 +78,8 @@
return nil
}
-// NewNNGConsumer create from deliver Mode
-func NewNNGConsumer(m Mode, url string, args ...interface{}) *NNG {
+// nngConsumer create from deliver Mode
+func nngConsumer(m Mode, url string, args ...interface{}) *NNG {
if sock, err := newSocket(protoConsumer(m)); err == nil {
if err = setSocketOptions(sock, args); err != nil {
@@ -92,12 +99,15 @@
return nil
}
-// MaxRecvSize max recv size
-var MaxRecvSize = 33 * 1024 * 1024
+// maxRecvSize max recv size
+var (
+ maxRecvSize = 33 * 1024 * 1024
+ surveyorTime = time.Second / 2
+)
func defualtSocketOptions(sock mangos.Socket) error {
var err error
- if err = sock.SetOption(mangos.OptionMaxRecvSize, MaxRecvSize); err != nil {
+ if err = sock.SetOption(mangos.OptionMaxRecvSize, maxRecvSize); err != nil {
sock.Close()
return err
}
@@ -127,7 +137,8 @@
if err != nil {
return err
}
- if sock.GetProtocol().Number() == mangos.ProtoSub {
+ switch sock.GetProtocol().Number() {
+ case mangos.ProtoSub:
for _, arg := range args {
switch arg.(type) {
case string:
@@ -136,6 +147,17 @@
err = sock.SetOption(mangos.OptionSubscribe, []byte(""))
}
}
+ case mangos.ProtoSurveyor:
+ for _, arg := range args {
+ switch arg.(type) {
+ case int:
+ surveyorTime = time.Duration(arg.(int)/2) * time.Second
+ default:
+ }
+ err = sock.SetOption(mangos.OptionSurveyTime, surveyorTime)
+ }
+ default:
+ fmt.Println("no additional args")
}
return nil
--
Gitblit v1.8.0