func replyLoop(sock mangos.Socket, done chan struct{}) {
	defer close(done)
	if sendData == nil {
		fatalf("No data to send!")
	}
	for {
		msg, err := sock.RecvMsg()
		switch err {
		case mangos.ErrRecvTimeout:
			return
		case nil:
		default:
			fatalf("RecvMsg failed: %v", err)
		}
		printMsg(msg)
		msg.Free()

		msg = mangos.NewMessage(len(sendData))
		msg.Body = append(msg.Body, sendData...)
		err = sock.SendMsg(msg)

		if err != nil {
			fatalf("SendMsg failed: %v", err)
		}
	}
}

func (w *wsPipe) Recv() (*mangos.Message, error) {

	// We ignore the message type for receive.
	_, body, err := w.ws.ReadMessage()
	if err != nil {
		return nil, err
	}
	msg := mangos.NewMessage(0)
	msg.Body = body
	return msg, nil
}

func (w *wsPipe) Recv() (*mangos.Message, error) {

	var buf []byte

	if err := websocket.Message.Receive(w.ws, &buf); err != nil {
		return nil, err
	}
	msg := mangos.NewMessage(len(buf))

	// This is kind of suboptimal copying...
	msg.Body = append(msg.Body, buf...)
	return msg, nil
}

// LatencyClient is the client side of the latency test.  It measures round
// trip times, and is the equivalent to nanomsg/perf/remote_lat.
func LatencyClient(addr string, msgSize int, roundTrips int) {
	s, err := pair.NewSocket()
	if err != nil {
		log.Fatalf("Failed to make new pair socket: %v", err)
	}
	defer s.Close()

	all.AddTransports(s)
	d, err := s.NewDialer(addr, nil)
	if err != nil {
		log.Fatalf("Failed to make new dialer: %v", err)
	}

	// TCP no delay, please!
	d.SetOption(mangos.OptionNoDelay, true)

	// Make sure we linger a bit on close...
	err = s.SetOption(mangos.OptionLinger, time.Second)
	if err != nil {
		log.Fatalf("Failed set Linger: %v", err)
	}

	err = d.Dial()
	if err != nil {
		log.Fatalf("Failed to dial: %v", err)
	}

	// 100 milliseconds to give TCP a chance to establish
	//time.Sleep(time.Millisecond * 100)
	msg := mangos.NewMessage(msgSize)
	msg.Body = msg.Body[0:msgSize]

	start := time.Now()
	for i := 0; i < roundTrips; i++ {
		if err = s.SendMsg(msg); err != nil {
			log.Fatalf("Failed SendMsg: %v", err)
		}
		if msg, err = s.RecvMsg(); err != nil {
			log.Fatalf("Failed RecvMsg: %v", err)
		}
	}
	finish := time.Now()
	msg.Free()

	total := (finish.Sub(start)) / time.Microsecond
	lat := float64(total) / float64(roundTrips*2)
	fmt.Printf("message size: %d [B]\n", msgSize)
	fmt.Printf("round trip count: %d\n", roundTrips)
	fmt.Printf("average latency: %.3f [us]\n", lat)
}

func starTestSender(t *testing.T, bt *starTester, cnt int) {
	defer close(bt.sdoneq)
	for i := 0; i < cnt; i++ {
		// Inject a small delay to give receivers a chance to catch up
		// Maximum is 10 msec.
		d := time.Duration(rand.Uint32() % 10000)
		time.Sleep(d * time.Microsecond)
		t.Logf("Peer %d: Sending %d", bt.id, i)
		msg := mangos.NewMessage(2)
		msg.Body = append(msg.Body, byte(bt.id), byte(i))
		if err := bt.sock.SendMsg(msg); err != nil {
			t.Errorf("Peer %d send %d fail: %v", bt.id, i, err)
			return
		}
	}
}

func (w *wsPipe) Recv() (*mangos.Message, error) {

	var buf []byte

	// prevent interleaved reads
	w.rlock.Lock()
	defer w.rlock.Unlock()

	if err := websocket.Message.Receive(w.ws, &buf); err != nil {
		return nil, err
	}
	msg := mangos.NewMessage(len(buf))

	// This is kind of suboptimal copying...
	msg.Body = append(msg.Body, buf...)
	return msg, nil
}

func (p *inproc) Send(m *mangos.Message) error {

	if p.peer == nil {
		return mangos.ErrClosed
	}

	// Upper protocols expect to have to pick header and body part.
	// Also we need to have a fresh copy of the message for receiver, to
	// break ownership.
	nmsg := mangos.NewMessage(len(m.Header) + len(m.Body))
	nmsg.Body = append(nmsg.Body, m.Header...)
	nmsg.Body = append(nmsg.Body, m.Body...)
	select {
	case p.wq <- nmsg:
		return nil
	case <-p.closeq:
		nmsg.Free()
		return mangos.ErrClosed
	}
}

func sendLoop(sock mangos.Socket) {
	if sendData == nil {
		fatalf("No data to send!")
	}
	for {
		msg := mangos.NewMessage(len(sendData))
		msg.Body = append(msg.Body, sendData...)
		err := sock.SendMsg(msg)

		if err != nil {
			fatalf("SendMsg failed: %v", err)
		}

		if sendInterval >= 0 {
			time.Sleep(time.Duration(sendInterval) * time.Second)
		} else {
			break
		}
	}
}

// Send sends a message.  For AF_SP_RAW messages the header must be
// included in the argument.  At this time, no flags are supported.
func (s *Socket) Send(b []byte, flags int) (int, error) {

	if flags != 0 {
		return -1, errNoFlag
	}

	m := mangos.NewMessage(len(b))
	m.Body = append(m.Body, b...)

	// Legacy nanomsg uses the opposite semantic for negative and
	// zero values than mangos.  A bit unfortunate.
	switch {
	case s.sto > 0:
		s.sock.SetOption(mangos.OptionSendDeadline, s.sto)
	case s.sto == 0:
		s.sock.SetOption(mangos.OptionSendDeadline, -1)
	case s.sto < 0:
		s.sock.SetOption(mangos.OptionSendDeadline, 0)
	}

	return len(b), s.sock.SendMsg(m)
}

func clientWorker(url string, id int) {
	var sock mangos.Socket
	var m *mangos.Message
	var err error

	if sock, err = req.NewSocket(); err != nil {
		die("can't get new req socket: %s", err.Error())
	}

	// Leave this in Cooked mode!

	sock.AddTransport(ipc.NewTransport())
	sock.AddTransport(tcp.NewTransport())
	if err = sock.Dial(url); err != nil {
		die("can't dial on req socket: %s", err.Error())
	}

	// send an empty messsage
	m = mangos.NewMessage(1)
	if err = sock.SendMsg(m); err != nil {
		die("can't send request: %s", err.Error())
	}

	if m, err = sock.RecvMsg(); err != nil {
		die("can't recv reply: %s", err.Error())
	}
	sock.Close()

	if len(m.Body) != 4 {
		die("bad response len: %d", len(m.Body))
	}

	worker := binary.BigEndian.Uint32(m.Body[0:])

	L.Lock()
	fmt.Printf("Client: %4d   Server: %4d\n", id, worker)
	L.Unlock()
}

func sendRecvLoop(sock mangos.Socket) {
	for {
		msg := mangos.NewMessage(len(sendData))
		msg.Body = append(msg.Body, sendData...)
		err := sock.SendMsg(msg)

		if err != nil {
			fatalf("SendMsg failed: %v", err)
		}

		if sendInterval < 0 {
			recvLoop(sock)
			return
		}

		now := time.Now()

		// maximum wait time is upper bound of recvTimeout and
		// sendInterval

		if recvTimeout < 0 || recvTimeout > sendInterval {
			sock.SetOption(mangos.OptionRecvDeadline,
				time.Second*time.Duration(sendInterval))
		}
		msg, err = sock.RecvMsg()
		switch err {
		case mangos.ErrRecvTimeout:
		case nil:
			printMsg(msg)
			msg.Free()
		default:
			fatalf("RecvMsg failed: %v", err)
		}
		time.Sleep((time.Second * time.Duration(sendInterval)) -
			time.Now().Sub(now))
	}
}

func TestTCPSendRecv(t *testing.T) {
	addr := "tcp://127.0.0.1:3333"
	ping := []byte("REQUEST_MESSAGE")
	ack := []byte("RESPONSE_MESSAGE")

	ch := make(chan *mangos.Message)

	t.Logf("Establishing listener")
	l, err := tran.NewListener(addr, protoRep)
	if err != nil {
		t.Errorf("NewListener failed: %v", err)
		return
	}
	defer l.Close()
	if err = l.Listen(); err != nil {
		t.Errorf("Listen failed: %v", err)
		return
	}

	go func() {
		defer close(ch)

		// Client side
		t.Logf("Connecting")
		d, err := tran.NewDialer(addr, protoReq)

		client, err := d.Dial()
		if err != nil {
			t.Errorf("Dial failed: %v", err)
			return
		}
		t.Logf("Connected client: %t", client.IsOpen())
		defer client.Close()

		req := mangos.NewMessage(len(ping))
		req.Body = append(req.Body, ping...)

		// Now try to send data
		t.Logf("Sending %d bytes", len(req.Body))

		err = client.Send(req)
		if err != nil {
			t.Errorf("Client send error: %v", err)
			return
		}
		t.Logf("Client sent")

		rep, err := client.Recv()
		if err != nil {
			t.Errorf("Client receive error: %v", err)
			return
		}

		if !bytes.Equal(rep.Body, ack) {
			t.Errorf("Reply mismatch: %v, %v", rep.Body, ack)
			return
		}
		if len(rep.Header) != 0 {
			t.Errorf("Client reply non-empty header: %v",
				rep.Header)
			return
		}
		select {
		case ch <- rep:
			t.Log("Client reply forwarded")
		case <-time.After(5 * time.Second): // 5 secs should be plenty
			t.Error("Client timeout forwarding reply")
			return
		}
	}()

	server, err := l.Accept()
	if err != nil {
		t.Errorf("Accept failed: %v", err)
		return
	}
	t.Logf("Connected server: %t", server.IsOpen())
	defer server.Close()

	// Now we can try to send and receive
	req, err := server.Recv()
	if err != nil {
		t.Errorf("Server receive error: %v", err)
		return
	}
	t.Logf("Server received %d bytes", len(req.Body))
	if !bytes.Equal(req.Body, ping) {
		t.Errorf("Request mismatch: %v, %v", req.Body, ping)
		return
	}

	if len(req.Header) != 0 {
		t.Errorf("Server request non-empty header: %v", req.Header)
		return
	}

	// Now reply
	rep := mangos.NewMessage(len(ack))
	rep.Body = append(rep.Body, ack...)

//.........这里部分代码省略.........

// MakeStart makes a start message, storing a 32-bit ID in the body.
func MakeStart(v uint32) *mangos.Message {
	m := mangos.NewMessage(10)
	m.Body = append(m.Body, []byte("START")...)
	m.Body = append(m.Body, byte(v>>24), byte(v>>16), byte(v>>8), byte(v))
	return m
}

func (c *T) NewMessage() *mangos.Message {
	return mangos.NewMessage(c.MsgSize)
}

func benchmarkPair(t *testing.B, url string, size int) {

	if strings.HasPrefix(url, "ipc://") && runtime.GOOS == "windows" {
		t.Skip("IPC not supported on Windows")
		return
	}

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

	if strings.HasPrefix(url, "wss://") || strings.HasPrefix(url, "tls+tcp://") {
		srvopts[mangos.OptionTLSConfig] = srvCfg
		cliopts[mangos.OptionTLSConfig] = cliCfg
	}

	finish := make(chan struct{})
	ready := make(chan struct{})
	srvsock, err := pair.NewSocket()
	if err != nil || srvsock == nil {
		t.Errorf("Failed creating server socket: %v", err)
		return
	}
	all.AddTransports(srvsock)
	defer srvsock.Close()

	clisock, err := pair.NewSocket()
	if err != nil || clisock == nil {
		t.Errorf("Failed creating client socket: %v", err)
		return
	}
	all.AddTransports(clisock)
	defer clisock.Close()

	go func() {
		var err error
		var m *mangos.Message

		if err = srvsock.ListenOptions(url, srvopts); err != nil {
			t.Errorf("Server listen failed: %v", err)
			return
		}
		close(ready)
		for i := 0; i < t.N; i++ {
			if m, err = srvsock.RecvMsg(); err != nil {
				t.Errorf("Error receiving %d: %v", i, err)
				return
			}
			m.Free()
		}
		close(finish)

	}()
	<-ready

	if err = clisock.DialOptions(url, cliopts); err != nil {
		t.Errorf("Client dial failed: %v", err)
		return
	}

	time.Sleep(700 * time.Millisecond)
	t.ResetTimer()

	for i := 0; i < t.N; i++ {
		msg := mangos.NewMessage(size)
		if err = clisock.SendMsg(msg); err != nil {
			t.Errorf("Client send failed: %v", err)
			return
		}
	}
	<-finish
	t.StopTimer()
	if size > 128 {
		t.SetBytes(int64(size))
	}
}