[zmq] zeromq snippets #MQ

main.go

func main() {
	if len(os.Args) < 2 || os.Args[1] != "client" {
		for i := 0; i < 4; i++ {
			go worker()
		}
		broker()
	} else {
		client()
	}
}

pirate_client.go

func client() {
	ctx, _ := zmq.NewContext()
	soc, _ := ctx.NewSocket(zmq.REQ)
	retries := 3
	poller := zmq.NewPoller()

	soc.Connect("tcp://localhost:8081")
	for retries > 0 {
		id := poller.Add(soc, zmq.POLLIN)
      
      	// send
		events, err := poller.Poll(3 * time.Second)
		if err != nil {
			log.Fatal(err)
		}
		if len(events) == 0 {
			retries--
			if retries == 0 {
				log.Fatal("server seems to be down, bail out")
			} else {
				log.Printf("no response from server, retrying...")
				soc.Close()

				poller.Remove(id)
				soc, _ = ctx.NewSocket(zmq.REQ)
				soc.Connect("tcp://localhost:8081")
			}
		} else {
			// receive
			poller.Remove(id)
			retries = 3
		}
	}
}

proxy_rd.go

func dealer() {
	ctx, _ := zmq.NewContext()
	frontend, _ := ctx.NewSocket(zmq.ROUTER)
	backend, _ := ctx.NewSocket(zmq.DEALER)
	poller := zmq.NewPoller()

	frontend.Bind("tcp://*:8081")
	backend.Bind("tcp://*:8080")
	poller.Add(frontend, zmq.POLLIN)
	poller.Add(backend, zmq.POLLIN)
	for {
		sockets, _ := poller.Poll(-1)
		for _, soc := range sockets {
			switch soc.Socket {
			case frontend:
			case backend:
			}
		}
	}
}