//  Echo service
func echo(socket *zmq.Socket) (err error) {
	msg, err := socket.RecvMessage(0)
	if err != nil {
		return
	}
	_, err = socket.SendMessage(msg)
	return
}

//  Application wants to speak to us, see if it's possible
func (bstar *Bstar) voter_ready(socket *zmq.Socket) error {
	//  If server can accept input now, call appl handler
	bstar.event = client_REQUEST
	err := bstar.execute_fsm()
	if err == nil {
		bstar.voter_fn(socket)
	} else {
		//  Destroy waiting message, no-one to read it
		socket.RecvMessage(0)
	}
	return nil
}

//  The RecvKvmsg function reads a key-value message from socket, and returns a new
//  Kvmsg instance.
func RecvKvmsg(socket *zmq.Socket) (kvmsg *Kvmsg, err error) {
	kvmsg = &Kvmsg{
		present: make([]bool, kvmsg_FRAMES),
		frame:   make([]string, kvmsg_FRAMES),
	}
	msg, err := socket.RecvMessage(0)
	if err != nil {
		return
	}
	//fmt.Printf("Recv from %s: %q\n", socket, msg)
	for i := 0; i < kvmsg_FRAMES && i < len(msg); i++ {
		kvmsg.frame[i] = msg[i]
		kvmsg.present[i] = true
	}
	return
}

func snapshots(socket *zmq.Socket, srv *clonesrv_t) (err error) {

	msg, err := socket.RecvMessage(0)
	if err != nil {
		return
	}
	identity := msg[0]

	//  Request is in second frame of message
	request := msg[1]
	if request != "ICANHAZ?" {
		err = errors.New("E: bad request, aborting")
		return
	}
	subtree := msg[2]

	//  Send state socket to client
	for _, kvmsg := range srv.kvmap {
		if key, _ := kvmsg.GetKey(); strings.HasPrefix(key, subtree) {
			socket.Send(identity, zmq.SNDMORE)
			kvmsg.Send(socket)
		}
	}

	//  Now send END message with sequence number
	log.Println("I: sending shapshot =", srv.sequence)
	socket.Send(identity, zmq.SNDMORE)
	kvmsg := kvmsg.NewKvmsg(srv.sequence)
	kvmsg.SetKey("KTHXBAI")
	kvmsg.SetBody(subtree)
	kvmsg.Send(socket)

	return
}

func NewPubsubClient(config PubsubConfig) (*PubsubClient, error) {

	context, err := zmq4.NewContext()
	var pub *zmq4.Socket
	var sub *zmq4.Socket

	if err != nil {
		return nil, err
	}

	if sub, err = context.NewSocket(zmq4.SUB); err != nil {
		return nil, err
	}

	if pub, err = context.NewSocket(zmq4.PUSH); err != nil {
		return nil, err
	}

	if err = pub.Connect(config.PubAddress); err != nil {

		return nil, err
	}

	if err = sub.Connect(config.SubAddress); err != nil {
		return nil, err
	}

	return &PubsubClient{pub, sub}, nil

}

func dump(soc *zmq.Socket) {
	fmt.Println("----------------------------------------")
	for {
		//  Process all parts of the message
		message, _ := soc.Recv(0)

		//  Dump the message as text or binary
		fmt.Printf("[%03d] ", len(message))
		if all_char.MatchString(message) {
			fmt.Print(message)
		} else {
			for i := 0; i < len(message); i++ {
				fmt.Printf("%02X ", message[i])
			}
		}
		fmt.Println()

		more, _ := soc.GetRcvmore()
		if !more {
			break
		}
	}
}

func Example_test_hwm() {

	MAX_SENDS := 10000
	BIND_FIRST := 1
	CONNECT_FIRST := 2

	test_defaults := func() int {

		// Set up bind socket
		bind_socket, err := zmq.NewSocket(zmq.PULL)
		if checkErr(err) {
			return 0
		}
		defer func() {
			err := bind_socket.Close()
			checkErr(err)
		}()

		err = bind_socket.Bind("inproc://a")
		if checkErr(err) {
			return 0
		}

		// Set up connect socket
		connect_socket, err := zmq.NewSocket(zmq.PUSH)
		if checkErr(err) {
			return 0
		}
		defer func() {
			err := connect_socket.Close()
			checkErr(err)
		}()

		err = connect_socket.Connect("inproc://a")
		if checkErr(err) {
			return 0
		}

		// Send until we block
		send_count := 0
		for send_count < MAX_SENDS {
			_, err := connect_socket.Send("", zmq.DONTWAIT)
			if err != nil {
				break
			}
			send_count++
		}

		// Now receive all sent messages
		recv_count := 0
		for {
			_, err := bind_socket.Recv(zmq.DONTWAIT)
			if err != nil {
				break
			}
			recv_count++
		}
		fmt.Println("send_count == recv_count:", send_count == recv_count)

		return send_count
	}

	count_msg := func(send_hwm, recv_hwm, testType int) int {

		var bind_socket, connect_socket *zmq.Socket
		var err error

		if testType == BIND_FIRST {
			// Set up bind socket
			bind_socket, err = zmq.NewSocket(zmq.PULL)
			if checkErr(err) {
				return 0
			}
			defer func() {
				err := bind_socket.Close()
				checkErr(err)
			}()

			err = bind_socket.SetRcvhwm(recv_hwm)
			if checkErr(err) {
				return 0
			}

			err = bind_socket.Bind("inproc://a")
			if checkErr(err) {
				return 0
			}

			// Set up connect socket
			connect_socket, err = zmq.NewSocket(zmq.PUSH)
			if checkErr(err) {
				return 0
			}
			defer func() {
				err := connect_socket.Close()
				checkErr(err)
			}()

			err = connect_socket.SetSndhwm(send_hwm)
			if checkErr(err) {
//.........这里部分代码省略.........

func bounce(server, client *zmq.Socket, willfail bool) {

	content := "12345678ABCDEFGH12345678abcdefgh"

	//  Send message from client to server
	rc, err := client.Send(content, zmq.SNDMORE|zmq.DONTWAIT)
	if checkErr0(err, 1) {
		return
	}
	if rc != 32 {
		checkErr0(errors.New("rc != 32"), 2)
	}

	rc, err = client.Send(content, zmq.DONTWAIT)
	if checkErr0(err, 3) {
		return
	}
	if rc != 32 {
		checkErr0(errors.New("rc != 32"), 4)
	}

	//  Receive message at server side
	msg, err := server.Recv(0)
	if checkErr0(e(err, willfail), 5) {
		return
	}

	//  Check that message is still the same
	if msg != content {
		checkErr0(errors.New(fmt.Sprintf("%q != %q", msg, content)), 6)
	}

	rcvmore, err := server.GetRcvmore()
	if checkErr0(err, 7) {
		return
	}
	if !rcvmore {
		checkErr0(errors.New(fmt.Sprint("rcvmore ==", rcvmore)), 8)
		return
	}

	//  Receive message at server side
	msg, err = server.Recv(0)
	if checkErr0(err, 9) {
		return
	}

	//  Check that message is still the same
	if msg != content {
		checkErr0(errors.New(fmt.Sprintf("%q != %q", msg, content)), 10)
	}

	rcvmore, err = server.GetRcvmore()
	if checkErr0(err, 11) {
		return
	}
	if rcvmore {
		checkErr0(errors.New(fmt.Sprint("rcvmore == ", rcvmore)), 12)
		return
	}

	// The same, from server back to client

	//  Send message from server to client
	rc, err = server.Send(content, zmq.SNDMORE)
	if checkErr0(err, 13) {
		return
	}
	if rc != 32 {
		checkErr0(errors.New("rc != 32"), 14)
	}

	rc, err = server.Send(content, 0)
	if checkErr0(err, 15) {
		return
	}
	if rc != 32 {
		checkErr0(errors.New("rc != 32"), 16)
	}

	//  Receive message at client side
	msg, err = client.Recv(0)
	if checkErr0(err, 17) {
		return
	}

	//  Check that message is still the same
	if msg != content {
		checkErr0(errors.New(fmt.Sprintf("%q != %q", msg, content)), 18)
	}

	rcvmore, err = client.GetRcvmore()
	if checkErr0(err, 19) {
		return
	}
	if !rcvmore {
		checkErr0(errors.New(fmt.Sprint("rcvmore ==", rcvmore)), 20)
		return
	}

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

func set_id(soc *zmq.Socket) {
	identity := fmt.Sprintf("%04X-%04X", rand.Intn(0x10000), rand.Intn(0x10000))
	soc.SetIdentity(identity)
}

func (server *server_t) ping(socket *zmq.Socket) {
	if time.Now().After(server.ping_at) {
		socket.SendMessage(server.endpoint, "PING")
		server.ping_at = time.Now().Add(PING_INTERVAL)
	}
}

//  Send key-value message to socket; any empty frames are sent as such.
func (kvmsg *Kvmsg) Send(socket *zmq.Socket) (err error) {
	//fmt.Printf("Send to %s: %q\n", socket, kvmsg.frame)
	kvmsg.encode_props()
	_, err = socket.SendMessage(kvmsg.frame)
	return
}