Golang github.com-alecthomas-gozmq.Socket类(方法)实例源码

func send(c chan string, socket *zmq.Socket) {
	for {
		msg := <-c
		socket.Send([]byte(msg), 0)
		fmt.Println("Pushed msg: ", msg)
	}
}

func send_and_recieve(socket zmq.Socket, in_msg []byte, id string) (res []byte, err error) {
	//		println("in_msg: ", string (in_msg))
	//	println("send ", id)
	var repeat bool
	var r0 []byte
	var err0 error

	repeat = true

	for repeat {

		socket.Send(in_msg, 0)
		//		println("ok")
		r0, err0 = socket.Recv(0)

		if r0 != nil && len(r0) == 3 {
			// это указание повторить запрос еще раз
			repeat = true
			time.Sleep(1e6)
		} else {
			repeat = false
		}
	}
	//	println("recv ", id)
	//			println("out_msg: ", string (r0))
	return r0, err0
}

func recv(c chan string, socket *zmq.Socket) {
	for {
		msg, _ := socket.Recv(0)
		fmt.Println("Pulled msg: ", string(msg))
		c <- string(msg)
	}
}

func zmqReceiver(s *zmq.Socket, c chan []map[string]interface{}, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		buf, err := s.Recv(0)
		if err != nil {
			log.Println("zmq.Socket.Recv():", err)
			continue
		}

		var m []map[string]interface{}
		err = json.Unmarshal(buf, &m)
		if err != nil {
			log.Println("json.Unmarshal():", err)
			continue
		}

		if !check(m) {
			log.Println("Invalid Command:", m)
			continue
		}

		select {
		case toDeliver := <-c:
			toDeliver = append(toDeliver, m...)
			c <- toDeliver
		default:
			c <- m
		}
	}
}

// send a message to the zmq REQ socket
func sendMessage(reqsocket zmq.Socket, m Message) {
	var address = PUB_KEY + "." + m.Nick
	b, _ := json.Marshal(m)
	var content = b
	env := envelope{address, string(content)}
	e, _ := json.Marshal(env)
	reqsocket.Send([]byte(e), 0)
	// wait for a reply
	reqsocket.Recv(0)
}

func runZmqStream() {
	var context zmq.Context
	var socket zmq.Socket

	// connect to zmq
	var err error
	if context, err = zmq.NewContext(); err != nil {
		panic("No ZMQ Context?")
	}
	defer context.Close()

	if socket, err = context.NewSocket(zmq.SUB); err != nil {
		panic("No ZMQ Socket Outbound??")
	}
	defer socket.Close()
	socket.Connect("tcp://localhost:5558")
	socket.SetSockOptString(zmq.SUBSCRIBE, "")

	for {
		// block here, waiting for inbound requests
		msg, _ := socket.Recv(0)
		if len(msg) > 0 {
			parts := strings.Split(string(msg), "\n\n")
			process(parts[0])
		}
	}
}

func receiveZmqMessage(subsocket zmq.Socket, m *Message) error {
	// using zmq multi-part messages which will arrive
	// in pairs. the first of which we don't care about so we discard.
	address, _ := subsocket.Recv(0)
	content, _ := subsocket.Recv(0)
	if startswith(string(address), PUB_KEY) {
		// it's one that we sent out, so ignore it
		return errors.New("do not echo my own messages")
	}
	return json.Unmarshal([]byte(content), m)
}

// The main announcer loop. Receives incoming messages, spits them back out to
// any connected subscribers.
func announcerLoop(listener *gozmq.Socket, sender *gozmq.Socket) {

	for {
		// This blocks until something comes down the pipe.
		msg, listenRecvError := listener.Recv(0)
		if listenRecvError != nil {
			errorHandler("listenRecvError", listenRecvError.Error())
		}

		sender.Send(msg, 0)
	}
}

// Publishes stored events to event listeners.
//
// Pops previously stored messages off a channel and published them to a
// ZeroMQ socket.
func publishAllSavedEvents(toPublish chan eventstore.StoredEvent, evpub zmq.Socket) {
	msg := make(zMsg, 3)
	for stored := range toPublish {
		msg[0] = stored.Event.Stream
		msg[1] = stored.Id
		msg[2] = stored.Event.Data

		if err := evpub.SendMultipart(msg, 0); err != nil {
			log.Println(err)
		}
	}
}

func proxyRouting(status chan bool) {
	// intialize the zmq context.
	context, err := zmq.NewContext()
	if err != nil {
		status <- false
		log.Fatal("Intialize the zeromq context failure.\n")
	}
	defer context.Close()

	var subscriber, publisher *zmq.Socket
	subscriber, err = context.NewSocket(zmq.XSUB)
	if err != nil {
		status <- false
		log.Fatal("Intialize the subscriber failure.\n")
	}
	defer subscriber.Close()

	var (
		sub_address, pub_address = "*", "*"
		subPort, pubPort         = 6001, 6000
	)

	// Bind the subscriber
	address := fmt.Sprintf("tcp://%s:%v", sub_address, subPort)
	err = subscriber.Bind(address)
	if err != nil {
		status <- false
		log.Fatalf("Subscriber bind on the address %s failure\n", address)
	}
	log.Printf("Subscriber bind on the address %s.\n", address)

	publisher, err = context.NewSocket(zmq.XPUB)
	if err != nil {
		status <- false
		log.Fatal("Intialize the publisher failure.\n")
	}
	defer publisher.Close()

	// Bind the publisher
	address = fmt.Sprintf("tcp://%s:%v", pub_address, pubPort)
	err = publisher.Bind(address)
	if err != nil {
		status <- false
		log.Fatalf("Publisher bind on the address %s failure.\n", address)
	}
	log.Printf("Publisher bind on the address %s.\n", address)

	log.Println("Proxy successfully launched...")
	// Poll the events on relevant sockets.
	zmq.Proxy(subscriber, publisher, nil)
}

func bind_to_channel(sock zmq.Socket) (channel chan []byte) {
	channel = make(chan []byte)
	go func() {
		for {
			msg, err := sock.Recv(0)
			if err != nil {
				fmt.Println("[ERROR] die at sock.Recv:", err.Error())
				break
			}
			channel <- msg
		}
	}()
	return
}

// A blocking function that will infinitely forward multi-part messages between two zmq.Sockets
func Forward(a, b zmq.Socket) {
	for {
		parts, err := a.RecvMultipart(0)
		if err != nil {
			log.Println("Error receiving message on frontend broker", err)
		}

		err = b.SendMultipart(parts, 0)
		if err != nil {
			log.Println("Error sending message on backend broker", err)
		}

		// log.Println("Brokered message:", Stringify(parts))
	}
}

func (t *TransportZmq) processControlIn(bridge_out *zmq.Socket) (ok bool) {
	var err error

RetryControl:
	msg, err := bridge_out.Recv(zmq.DONTWAIT)
	if err != nil {
		switch err {
		case syscall.EINTR:
			// Try again
			goto RetryControl
		case syscall.EAGAIN:
			// Poll lied, poll again
			return true
		}

		// Failure
		t.recv_chan <- fmt.Errorf("Pull zmq.Socket.Recv failure %s", err)
		return
	}

	switch string(msg) {
	case zmq_signal_output:
		// Start polling for send
		t.poll_items[1].Events = t.poll_items[1].Events | zmq.POLLOUT
	case zmq_signal_input:
		// If we staged a receive, process that
		if t.recv_buff != nil {
			select {
			case t.recv_bridge_chan <- t.recv_buff:
				t.recv_buff = nil

				// Start polling for receive
				t.poll_items[1].Events = t.poll_items[1].Events | zmq.POLLIN
			default:
				// Do nothing, we were asked for receive but channel is already full
			}
		} else {
			// Start polling for receive
			t.poll_items[1].Events = t.poll_items[1].Events | zmq.POLLIN
		}
	case zmq_signal_shutdown:
		// Shutdown
		return
	}

	ok = true
	return
}

func zmqSender(s *zmq.Socket, c chan []map[string]interface{}, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		fmt.Println("waiting channel")
		buf := <-c
		data, err := json.Marshal(buf)
		if err != nil {
			log.Println("json.Marshal():", err)
			continue
		}

		fmt.Println("Send String: " + string(data))
		s.Send(data, zmq.NOBLOCK)
	}
}

func dump(sink zmq.Socket) {
	parts, err := sink.RecvMultipart(0)
	if err != nil {
		fmt.Println(err)
	}
	for _, msgdata := range parts {
		is_text := true
		fmt.Printf("[%03d] ", len(msgdata))
		for _, char := range msgdata {
			if char < 32 || char > 127 {
				is_text = false
			}
		}
		if is_text {
			fmt.Printf("%s\n", msgdata)
		} else {
			fmt.Printf("%X\n", msgdata)
		}
	}
}

// SendResponse sends a message back to return identites of the received message.
func (receipt *MsgReceipt) SendResponse(socket *zmq.Socket, msg ComposedMsg) {
	socket.SendMultipart(receipt.Identities, zmq.SNDMORE)
	socket.Send([]byte("<IDS|MSG>"), zmq.SNDMORE)
	socket.SendMultipart(msg.ToWireMsg(receipt.Sockets.Key), 0)
	logger.Println("<--", msg.Header.Msg_type)
	logger.Printf("%+v\n", msg.Content)
}

// The core ZeroMQ messaging loop. Handles requests and responses
// asynchronously using the router socket. Every request is delegated to
// a goroutine for maximum concurrency.
//
// `gozmq` does currently not support copy-free messages/frames. This
// means that every message passing through this function needs to be
// copied in-memory. If this becomes a bottleneck in the future,
// multiple router sockets can be hooked to this final router to scale
// message copying.
//
// TODO: Make this a type function of `Server` to remove a lot of
// parameters.
func loopServer(estore *eventstore.EventStore, evpubsock, frontend zmq.Socket,
	stop chan bool) {
	toPoll := zmq.PollItems{
		zmq.PollItem{Socket: &frontend, zmq.Events: zmq.POLLIN},
	}

	pubchan := make(chan eventstore.StoredEvent)
	estore.RegisterPublishedEventsChannel(pubchan)
	go publishAllSavedEvents(pubchan, evpubsock)
	defer close(pubchan)

	pollchan := make(chan zmqPollResult)
	respchan := make(chan zMsg)

	pollCancel := make(chan bool)
	defer stopPoller(pollCancel)

	go asyncPoll(pollchan, toPoll, pollCancel)
	for {
		select {
		case res := <-pollchan:
			if res.err != nil {
				log.Println("Could not poll:", res.err)
			}
			if res.err == nil && toPoll[0].REvents&zmq.POLLIN != 0 {
				msg, _ := toPoll[0].Socket.RecvMultipart(0)
				zmsg := zMsg(msg)
				go handleRequest(respchan, estore, zmsg)
			}
			go asyncPoll(pollchan, toPoll, pollCancel)
		case frames := <-respchan:
			if err := frontend.SendMultipart(frames, 0); err != nil {
				log.Println(err)
			}
		case <-stop:
			log.Println("Server asked to stop. Stopping...")
			return
		}
	}
}

// ReadPb sends any protobuf along a ZMQ socket. This makes sure to bundle our
// type identifier at the beginning of the message.
func ReadPb(sock *zmq.Socket, timeout int) ([]byte, interface{}, error) {
	if timeout > 0 {
		if !WaitForRecv(sock, timeout) {
			return nil, nil, errors.New("recv timeout")
		}
	}

	rresp, err := sock.RecvMultipart(0)
	if err != nil {
		return nil, nil, err
	}

	// If we got a remote address, keep it.
	var remote []byte
	if len(rresp) > 1 {
		remote = rresp[0] // Remote address.
	}
	resp := rresp[len(rresp)-1]

	var pb interface{}
	switch resp[0] {
	case 1:
		pb = &Command{}
	case 2:
		pb = &StillAlive{}
	case 3:
		pb = &CommandFinished{}
	case 4:
		pb = &CommandOutput{}
	default:
		return nil, nil, errors.New(fmt.Sprintf("unknown packet type: %d", resp[0]))
	}

	err = proto.Unmarshal(resp[1:], pb.(proto.Message))
	if err != nil {
		return nil, nil, err
	}
	return remote, pb, nil
}

func (t *TransportZmq) bridge(bridge_in *zmq.Socket) {
	var message interface{}

	// Wait on channel, passing into socket
	// This keeps the socket in a single thread, otherwise we have to lock the entire publisher
	runtime.LockOSThread()

BridgeLoop:
	for {
		select {
		case notify := <-t.bridge_chan:
			bridge_in.Send(notify, 0)

			// Shutdown?
			if string(notify) == zmq_signal_shutdown {
				break BridgeLoop
			}
		case message = <-t.recv_bridge_chan:
			// The reason we flush recv through the bridge and not directly to recv_chan is so that if
			// the poller was quick and had to cache a receive as the channel was full, it will stop
			// polling - flushing through bridge allows us to signal poller to start polling again
			// It is not the publisher's responsibility to do this, and TLS wouldn't need it
			bridge_in.Send([]byte(zmq_signal_input), 0)

			// Keep trying to forward on the message
		ForwardLoop:
			for {
				select {
				case notify := <-t.bridge_chan:
					bridge_in.Send(notify, 0)

					// Shutdown?
					if string(notify) == zmq_signal_shutdown {
						break BridgeLoop
					}
				case t.recv_chan <- message:
					break ForwardLoop
				}
			}
		}
	}

	// We should linger by default to ensure shutdown is transmitted
	bridge_in.Close()
	runtime.UnlockOSThread()
	t.wait.Done()
}

// SendResponse sends a message back to return identites of the received message.
func (receipt *MsgReceipt) SendResponse(socket *zmq.Socket, msg ComposedMsg) {

	socket.SendMultipart(receipt.Identities, zmq.SNDMORE)
	socket.Send([]byte("<IDS|MSG>"), zmq.SNDMORE)

	msgParts, err := msg.ToWireMsg(receipt.Sockets.Key)
	if err != nil {
		log.Fatalln(err)
	}
	socket.SendMultipart(msgParts, 0)
	logger.Println("<--", msg.Header.MsgType)
	logger.Printf("%+v\n", msg.Content)
}