From 2d390df9ede39c9d7c34bd8190b9329cfc371325 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期五, 27 九月 2019 16:19:05 +0800
Subject: [PATCH] remove exist ipc

---
 nng.go |  213 +++++++++++++++++++++++++++--------------------------
 1 files changed, 109 insertions(+), 104 deletions(-)

diff --git a/nng.go b/nng.go
index 84d3699..812cc37 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"
@@ -23,122 +24,122 @@
 // NNG mangos wrap
 type NNG struct {
 	sock mangos.Socket
+	typ  td
+	mode Mode
+	url  string
+
+	arguments []interface{}
 }
 
 // Send impl interface Diliver
 func (n *NNG) Send(data []byte) error {
-	if n.sock == nil {
+	if n == nil {
 		return errors.New("please init NNG first")
 	}
-
-	if _, err := n.sock.GetOption(mangos.OptionRaw); err == nil {
-		msg := mangos.NewMessage(len(data))
-		msg.Body = data
-		return n.sock.SendMsg(msg)
+	var err error
+	if n.sock == nil {
+		n.sock, err = n.makeNNG(agent)
+		if err != nil {
+			fmt.Println("create nng sender error")
+			return err
+		}
 	}
-	return n.sock.Send(data)
+
+	if surveyorTime > 0 {
+		time.Sleep(time.Duration(surveyorTime*2) * time.Second)
+	}
+
+	msg := mangos.NewMessage(1)
+	msg.Body = data
+	return n.sock.SendMsg(msg)
 }
 
 // Recv impl interface Diliver
 func (n *NNG) Recv() ([]byte, error) {
-	if n.sock == nil {
+	if n == nil {
 		return nil, errors.New("please init NNG first")
 	}
-	if _, err := n.sock.GetOption(mangos.OptionRaw); err == nil {
-		msg, err := n.sock.RecvMsg()
-		return msg.Body, err
+
+	var err error
+
+	if n.sock == nil {
+		n.sock, err = n.makeNNG(coactee)
+		if err != nil {
+			fmt.Println("create nng reciever error")
+			return nil, err
+		}
 	}
-	return n.sock.Recv()
+
+	var msg *mangos.Message
+	if msg, err = n.sock.RecvMsg(); err != nil {
+		return nil, err
+	}
+	return msg.Body, nil
+
 }
 
-// nngProducer create from deliver Mode
-func nngProducer(m Mode, url string, args ...interface{}) *NNG {
+// 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 != nil && n.sock != nil {
+		n.sock.Close()
+		n.sock = nil
+	}
+}
+
+func nngServer(m Mode, url string, args ...interface{}) *NNG {
 
 	rmExistedIpcName(url)
-	if sock, err := newSocket(protoProducer(m)); err == nil {
-		if err = setSocketOptions(sock, args); err != nil {
-			return nil
-		}
-		if err = sock.Listen(url); err != nil {
-			sock.Close()
-			return nil
-		}
-		return &NNG{
-			sock,
-		}
-	}
 
-	return nil
+	return &NNG{
+		typ:       agent,
+		mode:      m,
+		url:       url,
+		arguments: args,
+	}
 }
 
-// nngConsumer create from deliver Mode
-func nngConsumer(m Mode, url string, args ...interface{}) *NNG {
+func nngClient(m Mode, url string, args ...interface{}) *NNG {
 
-	if sock, err := newSocket(protoConsumer(m)); err == nil {
-		if err = setSocketOptions(sock, args); err != nil {
-			return nil
-		}
-
-		if err = sock.Dial(url); err != nil {
-			sock.Close()
-			return nil
-		}
-
-		return &NNG{
-			sock,
-		}
+	return &NNG{
+		typ:       coactee,
+		mode:      m,
+		url:       url,
+		arguments: args,
 	}
 
-	return nil
 }
 
-// MaxRecvSize max recv size
-var MaxRecvSize = 33 * 1024 * 1024
+func proto(typ td, m Mode) protocol {
+	if typ == agent {
+		return protoAgent(m)
+	} else if typ == coactee {
+		return protoCoactee(m)
+	}
+	return NONE
+}
 
-func defualtSocketOptions(sock mangos.Socket) error {
+func (n *NNG) makeNNG(typ td) (mangos.Socket, error) {
+
+	var sock mangos.Socket
 	var err error
-	if err = sock.SetOption(mangos.OptionMaxRecvSize, MaxRecvSize); err != nil {
-		sock.Close()
-		return err
-	}
-	if err = sock.SetOption(mangos.OptionWriteQLen, 0); err != nil {
-		sock.Close()
-		return err
-	}
-	if err = sock.SetOption(mangos.OptionReadQLen, 0); err != nil {
-		sock.Close()
-		return err
-	}
-	// if err = sock.SetOption(mangos.OptionNoDelay, true); err != nil {
-	// 	sock.Close()
-	// 	return err
-	// }
-	if err = sock.SetOption(mangos.OptionRaw, true); err != nil {
-		sock.Close()
-		return err
+
+	switch n.mode {
+	case Bus:
+		sock, err = n.busMakeNNG(typ)
+	case ReqRep, SurvResp:
+		sock, err = n.rrMakeNNG(typ)
+	default:
+		sock, err = n.ppMakeNNG(typ)
 	}
 
-	return nil
-}
-
-func setSocketOptions(sock mangos.Socket, args ...interface{}) error {
-
-	err := defualtSocketOptions(sock)
-	if err != nil {
-		return err
-	}
-	if sock.GetProtocol().Number() == mangos.ProtoSub {
-		for _, arg := range args {
-			switch arg.(type) {
-			case string:
-				err = sock.SetOption(mangos.OptionSubscribe, []byte(arg.(string)))
-			default:
-				err = sock.SetOption(mangos.OptionSubscribe, []byte(""))
-			}
-		}
-	}
-
-	return nil
+	return sock, err
 }
 
 func rmExistedIpcName(url string) {
@@ -146,6 +147,8 @@
 
 	if s[0] == "ipc" {
 		if _, err := os.Stat(s[1]); err == nil {
+			os.Remove(s[1])
+		} else if !os.IsNotExist(err) {
 			os.Remove(s[1])
 		}
 	}
@@ -155,6 +158,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
 
@@ -188,8 +194,6 @@
 	}
 
 	all.AddTransports(s)
-	// s.AddTransport(ipc.NewTransport())
-	// s.AddTransport(tcp.NewTransport())
 
 	return s, nil
 }
@@ -205,6 +209,7 @@
 
 // Constants for protocols.
 const (
+	NONE       = -1
 	PUSH       = protocol(mangos.ProtoPush)
 	PULL       = protocol(mangos.ProtoPull)
 	PUB        = protocol(mangos.ProtoPub)
@@ -217,38 +222,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
 }

--
Gitblit v1.8.0