func main() {
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	backend, _ := zmq.NewSocket(zmq.ROUTER)
	defer frontend.Close()
	defer backend.Close()
	frontend.Bind("tcp://*:5555") //  For clients
	backend.Bind("tcp://*:5556")  //  For workers

	//  Queue of available workers
	workers := make([]string, 0)

	poller1 := zmq.NewPoller()
	poller1.Add(backend, zmq.POLLIN)
	poller2 := zmq.NewPoller()
	poller2.Add(backend, zmq.POLLIN)
	poller2.Add(frontend, zmq.POLLIN)

	//  The body of this example is exactly the same as lbbroker2.
LOOP:
	for {
		//  Poll frontend only if we have available workers
		var sockets []zmq.Polled
		var err error
		if len(workers) > 0 {
			sockets, err = poller2.Poll(-1)
		} else {
			sockets, err = poller1.Poll(-1)
		}
		if err != nil {
			break //  Interrupted
		}
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case backend: //  Handle worker activity on backend
				//  Use worker identity for load-balancing
				msg, err := s.RecvMessage(0)
				if err != nil {
					break LOOP //  Interrupted
				}
				var identity string
				identity, msg = unwrap(msg)
				workers = append(workers, identity)

				//  Forward message to client if it's not a READY
				if msg[0] != WORKER_READY {
					frontend.SendMessage(msg)
				}

			case frontend:
				//  Get client request, route to first available worker
				msg, err := s.RecvMessage(0)
				if err == nil {
					backend.SendMessage(workers[0], "", msg)
					workers = workers[1:]
				}
			}
		}
	}
}

//  Connect or reconnect to broker.
func (mdcli *Mdcli) ConnectToBroker() (err error) {
	if mdcli.client != nil {
		mdcli.client.Close()
		mdcli.client = nil
	}
	mdcli.client, err = zmq.NewSocket(zmq.REQ)
	if err != nil {
		if mdcli.verbose {
			log.Println("E: ConnectToBroker() creating socket failed")
		}
		return
	}
	mdcli.poller = zmq.NewPoller()
	mdcli.poller.Add(mdcli.client, zmq.POLLIN)

	if mdcli.verbose {
		log.Printf("I: connecting to broker at %s...", mdcli.broker)
	}
	err = mdcli.client.Connect(mdcli.broker)
	if err != nil && mdcli.verbose {
		log.Println("E: ConnectToBroker() failed to connect to broker", mdcli.broker)
	}

	return
}

func main() {

	//  Connect to task ventilator
	receiver, _ := zmq.NewSocket(zmq.PULL)
	defer receiver.Close()
	receiver.Connect("tcp://localhost:5557")

	//  Connect to weather server
	subscriber, _ := zmq.NewSocket(zmq.SUB)
	defer subscriber.Close()
	subscriber.Connect("tcp://localhost:5556")
	subscriber.SetSubscribe("10001 ")

	//  Initialize poll set
	poller := zmq.NewPoller()
	poller.Add(receiver, zmq.POLLIN)
	poller.Add(subscriber, zmq.POLLIN)
	//  Process messages from both sockets
	for {
		sockets, _ := poller.Poll(-1)
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case receiver:
				task, _ := s.Recv(0)
				//  Process task
				fmt.Println("Got task:", task)
			case subscriber:
				update, _ := s.Recv(0)
				//  Process weather update
				fmt.Println("Got weather update:", update)
			}
		}
	}
}

func (k *kernelRunner) loop() error {

	poller := zmq.NewPoller()
	poller.Add(k.sockets.Shell, zmq.POLLIN)
	poller.Add(k.sockets.Stdin, zmq.POLLIN)
	poller.Add(k.sockets.Control, zmq.POLLIN)

	for !k.shutdown {
		polled, err := poller.Poll(-1)
		if err != nil {
			return fmt.Errorf("poll failed: %v", err)
		}
		for _, polled := range polled {
			msg, ids, err := k.readMessage(polled.Socket)
			if err != nil {
				return fmt.Errorf("reading message: %v", err)
			}
			switch polled.Socket {
			case k.sockets.Shell, k.sockets.Control:
				err := k.handleShellOrControl(msg, ids, polled.Socket)
				if err != nil {
					log.Printf("handling request: %v", err)
				}
			case k.sockets.Stdin:
				if err := k.handleStdin(msg, ids); err != nil {
					log.Printf("handling stdin: %v", err)
				}
			}
			if k.shutdown {
				break
			}
		}
	}
	return nil
}

func (i *Incoming) listen() {
	poller := zmq4.NewPoller()
	poller.Add(i.skt, zmq4.POLLIN)

	for {
		i.m.Lock()
		if i.stopped {
			i.m.Unlock()

			return
		}
		sockets, err := poller.Poll(100 * time.Millisecond)
		if err != nil {
			continue
		}
		for range sockets {

			msg, err := i.skt.RecvMessage(0)
			if err == nil {
				i.in.Add(Message{i.addr, msg})
			}
		}
		i.m.Unlock()
	}
}

// ZmqPubSubProxy Implements a many-to-many device on a zmq PUB-SUB connection
func ZmqPubSubProxy(host string, portIn int, portOut int, logger *log.Logger) {
	xsub, _ := zmq.NewSocket(zmq.SUB)
	xpub, _ := zmq.NewSocket(zmq.PUB)

	defer xsub.Close()
	defer xpub.Close()

	addrIn := fmt.Sprintf("tcp://*:%d", portIn)
	addrOut := fmt.Sprintf("tcp://*:%d", portOut)

	logger.Println("ZMQ XSUB on", addrIn)
	xsub.Bind(addrIn)
	xsub.SetSubscribe("")

	logger.Println("ZMQ XPUB on", addrOut)
	xpub.Bind(addrOut)

	poller := zmq.NewPoller()
	poller.Add(xsub, zmq.POLLIN)

	for {
		// keep looping
		sockets, _ := poller.Poll(5 * time.Second)
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case xsub:
				ZmqSendMulti(xpub, ZmqRecvMulti(s))
			}
		}
	}
}

func (pcli *PigatoClient) ConnectToBroker() (err error) {
	if pcli.ctx.client != nil {
		pcli.ctx.client.Close()
		pcli.ctx.client = nil
	}
	pcli.ctx.client, err = zmq.NewSocket(zmq.DEALER)
	pcli.identity = randSeq()
	pcli.ctx.client.SetIdentity(pcli.identity)

	if err != nil {
		if pcli.verbose {
			log.Println("E: ConnectToBroker() creating socket failed")
		}
		return
	}
	pcli.ctx.poller = zmq.NewPoller()
	pcli.ctx.poller.Add(pcli.ctx.client, zmq.POLLIN)

	if pcli.verbose {
		log.Printf("I: connecting to broker at %s...", pcli.broker)
	}
	err = pcli.ctx.client.Connect(pcli.broker)
	if err != nil && pcli.verbose {
		log.Println("E: ConnectToBroker() failed to connect to broker", pcli.broker)
	}

	return
}

// run zmq event loop
func zmqMsgHandler() {
	subscriber := setupSocket()
	defer subscriber.Close()

	poller := zmq.NewPoller()
	poller.Add(subscriber, zmq.POLLIN)

	for !interrupted {
		sockets, _ := poller.Poll(1 * time.Second)
		for _, socket := range sockets {
			s := socket.Socket
			msg, _ := s.RecvMessage(0)
			if len(msg) != 2 {
				logError("got invalid message: %v", msg)
				continue
			}
			var app_env, data = msg[0], msg[1]
			var msgType int
			if strings.Contains(data, "total_time") {
				msgType = perfMsg
			} else {
				msgType = errorMsg
			}
			zmq_channel <- &ZmqMsg{msgType: msgType, app_env: app_env, data: data}
		}
	}
}

func main() {
	//  Initialize polling
	poller = zmq.NewPoller()

	//  Prepare our frontend sockets
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	defer frontend.Close()
	address = "tcp://*:5559"
	frontend.Bind(address)
	//  Initialize frontend poll set
	poller.Add(frontend, zmq.POLLIN)

	//  Load Service List
	getServiceList()
	//Make sure all services in list are closed at end of execution
	for _, service := range services {
		defer service.Backend.Close()
	}
	//  Initialize polling and binding to addresses
	initializeServices()

	fmt.Println("Broker at ", address, " waiting for connection...")
	//  List available services
	listServices()

	//  Switch messages between sockets
	for {
		sockets, _ := poller.Poll(-1)
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case frontend:
				backend := serveFrontend(s)
				//  (For newly registered services)
				//  Make sure socket is closed at end of execution
				if backend != nil {
					defer backend.Close()
					backend = nil
				}

			//  All services fall under default
			default:
				//fmt.Println("Receiving message from service...")
				for {
					msg, _ := s.Recv(0)
					//fmt.Printf("\tIn-->%s\n", msg)
					if more, _ := s.GetRcvmore(); more {
						//fmt.Printf("\tForwarding to client: %s\n", msg)
						frontend.Send(msg, zmq.SNDMORE)
					} else {
						//fmt.Printf("\tForwarding to client: %s\n", msg)
						frontend.Send(msg, 0)
						//fmt.Println("\tDone")
						break
					}
				} //end for (receive message from service)
			} //end switch (sockets polled)
		} //end for(sockets polled)
	} //end for(forever)
}

func main() {

	//config := ReadConfig("")
	//  Prepare our sockets
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	defer frontend.Close()
	backend, _ := zmq.NewSocket(zmq.DEALER)
	defer backend.Close()
	frontend.Bind(frontend_url_port)
	backend.Bind(backend_url_port)

	//  Initialize poll set
	poller := zmq.NewPoller()
	poller.Add(frontend, zmq.POLLIN)
	poller.Add(backend, zmq.POLLIN)

	log.Println("0MQ broker started!")
	log.Println("Frontend protocl/url/port:", frontend_url_port)
	log.Println("Backend protocol/url/port:", backend_url_port)

	//  Switch messages between sockets
	for {
		sockets, _ := poller.Poll(-1)
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case frontend:
				for {
					msg, _ := s.Recv(0)
					if more, _ := s.GetRcvmore(); more {
						backend.Send(msg, zmq.SNDMORE)
					} else {
						backend.Send(msg, 0)
						break
					}
					if debug {
						log.Printf("relayed request: [%s]\n", msg)
					}

				}
			case backend:
				for {
					msg, _ := s.Recv(0)
					if more, _ := s.GetRcvmore(); more {
						frontend.Send(msg, zmq.SNDMORE)
					} else {
						frontend.Send(msg, 0)
						break
					}
					if debug {
						log.Printf("relayed reply: [%s]\n", msg)
					}
				}

			}
		}
	}
	//log.Fatalln("Proxy interrupted:", err)
}

func flcliapi_agent() {

	agent := agent_new()

	poller := zmq.NewPoller()
	poller.Add(agent.pipe, zmq.POLLIN)
	poller.Add(agent.router, zmq.POLLIN)
	for {
		//  Calculate tickless timer, up to 1 hour
		tickless := time.Now().Add(time.Hour)
		if len(agent.request) > 0 && tickless.After(agent.expires) {
			tickless = agent.expires
		}
		for key := range agent.servers {
			tickless = agent.servers[key].tickless(tickless)
		}

		polled, err := poller.Poll(tickless.Sub(time.Now()))
		if err != nil {
			break //  Context has been shut down
		}

		for _, item := range polled {
			switch item.Socket {
			case agent.pipe:
				agent.control_message()
			case agent.router:
				agent.router_message()
			}
		}

		//  If we're processing a request, dispatch to next server
		if len(agent.request) > 0 {
			if time.Now().After(agent.expires) {
				//  Request expired, kill it
				agent.pipe.SendMessage("FAILED")
				agent.request = agent.request[0:0]
			} else {
				//  Find server to talk to, remove any expired ones
				for len(agent.actives) > 0 {
					server := agent.actives[0]
					if time.Now().After(server.expires) {
						agent.actives = agent.actives[1:]
						server.alive = false
					} else {
						agent.router.SendMessage(server.endpoint, agent.request)
						break
					}
				}
			}
		}
		//  --(Disconnect and delete any expired servers)--
		//  Send heartbeats to idle servers if needed
		for key := range agent.servers {
			agent.servers[key].ping(agent.router)
		}
	}
}

func main() {
	frontend, _ := zmq.NewSocket(zmq.SUB)
	frontend.Bind("tcp://*:5557")
	backend, _ := zmq.NewSocket(zmq.XPUB)
	backend.Bind("tcp://*:5558")

	//  Subscribe to every single topic from publisher
	frontend.SetSubscribe("")

	//  Store last instance of each topic in a cache
	cache := make(map[string]string)

	//  We route topic updates from frontend to backend, and
	//  we handle subscriptions by sending whatever we cached,
	//  if anything:
	poller := zmq.NewPoller()
	poller.Add(frontend, zmq.POLLIN)
	poller.Add(backend, zmq.POLLIN)
LOOP:
	for {
		polled, err := poller.Poll(1000 * time.Millisecond)
		if err != nil {
			break //  Interrupted
		}

		for _, item := range polled {
			switch socket := item.Socket; socket {
			case frontend:
				//  Any new topic data we cache and then forward
				msg, err := frontend.RecvMessage(0)
				if err != nil {
					break LOOP
				}
				cache[msg[0]] = msg[1]
				backend.SendMessage(msg)
			case backend:
				//  When we get a new subscription we pull data from the cache:
				msg, err := backend.RecvMessage(0)
				if err != nil {
					break LOOP
				}
				frame := msg[0]
				//  Event is one byte 0=unsub or 1=sub, followed by topic
				if frame[0] == 1 {
					topic := frame[1:]
					fmt.Println("Sending cached topic", topic)
					previous, ok := cache[topic]
					if ok {
						backend.SendMessage(topic, previous)
					}
				}
			}
		}
	}
}

func main() {
	verbose := false
	if len(os.Args) > 1 && os.Args[1] == "-v" {
		verbose = true
	}

	broker, _ := NewBroker(verbose)
	broker.Bind("tcp://*:5555")

	poller := zmq.NewPoller()
	poller.Add(broker.socket, zmq.POLLIN)

	//  Get and process messages forever or until interrupted
	for {
		polled, err := poller.Poll(HEARTBEAT_INTERVAL)
		if err != nil {
			break //  Interrupted
		}

		//  Process next input message, if any
		if len(polled) > 0 {
			msg, err := broker.socket.RecvMessage(0)
			if err != nil {
				break //  Interrupted
			}
			if broker.verbose {
				log.Printf("I: received message: %q\n", msg)
			}
			sender, msg := popStr(msg)
			_, msg = popStr(msg)
			header, msg := popStr(msg)

			switch header {
			case mdapi.MDPC_CLIENT:
				broker.ClientMsg(sender, msg)
			case mdapi.MDPW_WORKER:
				broker.WorkerMsg(sender, msg)
			default:
				log.Printf("E: invalid message: %q\n", msg)
			}
		}
		//  Disconnect and delete any expired workers
		//  Send heartbeats to idle workers if needed
		if time.Now().After(broker.heartbeat_at) {
			broker.Purge()
			for _, worker := range broker.waiting {
				worker.Send(mdapi.MDPW_HEARTBEAT, "", []string{})
			}
			broker.heartbeat_at = time.Now().Add(HEARTBEAT_INTERVAL)
		}
	}
	log.Println("W: interrupt received, shutting down...")
}

func request(t string, m string) (r string, err error) {
	sck, err := newSocket(zmq4.REQ, 5555)
	defer sck.Close()

	sck.SetLinger(0)

	if err != nil {
		err = ErrConnectionFailed
		return
	}

	if _, err = sck.Send(t+" "+m, 0); err != nil {
		err = ErrConnectionFailed
		return
	}

	poller := zmq4.NewPoller()
	poller.Add(sck, zmq4.POLLIN)

	sockets, err := poller.Poll(requestTimeout)

	if err != nil {
		err = ErrConnectionFailed
		return
	}

	if len(sockets) < 1 {
		err = ErrConnectionFailed
		return
	}

	r, err = sck.Recv(0)

	if err != nil {
		err = ErrConnectionFailed
		return
	}

	if strings.Contains(r, "ERROR") {
		parts := strings.Split(r, "|")

		if len(parts) > 1 {
			err = errors.New(parts[1])
		} else {
			err = errors.New("Unknown")
		}

		return
	}

	return
}

func s_worker_socket() (*zmq.Socket, *zmq.Poller) {
	worker, _ := zmq.NewSocket(zmq.DEALER)
	worker.Connect("tcp://localhost:5556")

	//  Tell queue we're ready for work
	fmt.Println("I: worker ready")
	worker.Send(PPP_READY, 0)

	poller := zmq.NewPoller()
	poller.Add(worker, zmq.POLLIN)

	return worker, poller
}

/* {{{ func (m *MQPool) Get(key string) (mq *msgqueue, err error) {
 * 获取相关key的队列
 */
func (m *MQPool) Get(key string) (mq *msgqueue, err error) {
	// hash key
	hk := fmt.Sprintf("%x", md5.Sum([]byte(key)))
	now := time.Now()
	expire := now.Add(m.life)
	if _, ok := m.Pool[hk]; ok {
		mq = m.Pool[hk]
	} else {
		if len(m.Pool) >= m.max { //达到最大数,清理
			for k, queue := range m.Pool {
				if now.After(queue.expire) { //过期,死亡
					m.remove(k)
				}
			}
		}
		if len(m.Pool) < m.max {
			mq = &msgqueue{
				pusher:  NewSocket(zmq.DEALER, 1000),
				queuer:  NewSocket(zmq.DEALER, 1000),
				iPoller: zmq.NewPoller(), //in
				oPoller: zmq.NewPoller(), //out
				expire:  expire,
			}
			//建立连接
			mq.pusher.Bind(fmt.Sprint("inproc://", hk))
			mq.queuer.Connect(fmt.Sprint("inproc://", hk))
			mq.oPoller.Add(mq.pusher.soc, zmq.POLLOUT)
			mq.iPoller.Add(mq.queuer.soc, zmq.POLLIN)
			m.Pool[hk] = mq
		} else {
			// pool 满了, 婉拒
			err = fmt.Errorf("pool space full: %d", m.max)
		}
	}
	if err == nil {
		mq.expire = expire
	}
	return
}

func main() {
	//  First argument is this broker's name
	//  Other arguments are our peers' names
	//
	if len(os.Args) < 2 {
		fmt.Println("syntax: peering1 me {you}...")
		os.Exit(1)
	}
	self := os.Args[1]
	fmt.Printf("I: preparing broker at %s...\n", self)
	rand.Seed(time.Now().UnixNano())

	//  Bind state backend to endpoint
	statebe, _ := zmq.NewSocket(zmq.PUB)
	defer statebe.Close()
	statebe.Bind("ipc://" + self + "-state.ipc")

	//  Connect statefe to all peers
	statefe, _ := zmq.NewSocket(zmq.SUB)
	defer statefe.Close()
	statefe.SetSubscribe("")
	for _, peer := range os.Args[2:] {
		fmt.Printf("I: connecting to state backend at '%s'\n", peer)
		statefe.Connect("ipc://" + peer + "-state.ipc")
	}

	//  The main loop sends out status messages to peers, and collects
	//  status messages back from peers. The zmq_poll timeout defines
	//  our own heartbeat:

	poller := zmq.NewPoller()
	poller.Add(statefe, zmq.POLLIN)
	for {
		//  Poll for activity, or 1 second timeout
		sockets, err := poller.Poll(time.Second)
		if err != nil {
			break
		}

		//  Handle incoming status messages
		if len(sockets) == 1 {
			msg, _ := statefe.RecvMessage(0)
			peer_name := msg[0]
			available := msg[1]
			fmt.Printf("%s - %s workers free\n", peer_name, available)
		} else {
			statebe.SendMessage(self, rand.Intn(10))
		}
	}
}

func New(addr string) (*Router, error) {
	ctx, err := zmq.NewContext()
	if err != nil {
		return nil, err
	}
	router := &Router{
		ctx:       ctx,
		receivers: make(map[*zmq.Socket]*receiver),
		poller:    zmq.NewPoller(),
		addr:      addr,
	}

	return router, nil
}

func main() {
	//  Socket to receive messages on
	receiver, _ := zmq.NewSocket(zmq.PULL)
	defer receiver.Close()
	receiver.Connect("tcp://localhost:5557")

	//  Socket to send messages to
	sender, _ := zmq.NewSocket(zmq.PUSH)
	defer sender.Close()
	sender.Connect("tcp://localhost:5558")

	//  Socket for control input
	controller, _ := zmq.NewSocket(zmq.SUB)
	defer controller.Close()
	controller.Connect("tcp://localhost:5559")
	controller.SetSubscribe("")

	//  Process messages from receiver and controller
	poller := zmq.NewPoller()
	poller.Add(receiver, zmq.POLLIN)
	poller.Add(controller, zmq.POLLIN)
	//  Process messages from both sockets
LOOP:
	for {
		sockets, _ := poller.Poll(-1)
		for _, socket := range sockets {
			switch s := socket.Socket; s {
			case receiver:
				msg, _ := s.Recv(0)

				//  Do the work
				t, _ := strconv.Atoi(msg)
				time.Sleep(time.Duration(t) * time.Millisecond)

				//  Send results to sink
				sender.Send(msg, 0)

				//  Simple progress indicator for the viewer
				fmt.Printf(".")
			case controller:
				//  Any controller command acts as 'KILL'
				break LOOP //  Exit loop
			}
		}
	}
	fmt.Println()
}

func BrokerStart(verbose bool) {
	broker, _ := NewBroker(verbose)
	broker.Bind("tcp://*:55555")

	poller := zmq.NewPoller()
	poller.Add(broker.socket, zmq.POLLIN)

	//  Get and process messages forever or until interrupted
	for {
		polled, err := poller.Poll(HEARTBEAT_INTERVAL)
		if err != nil {
			break //  Interrupted
		}

		//  Process next input message, if any
		if len(polled) > 0 {
			msg, err := broker.socket.RecvMessage(0)
			if err != nil {
				break //  Interrupted
			}
			if broker.verbose {
				log.Printf("I: received message: %q\n", msg)
			}
			sender, msg := pgtlib.PopStr(msg)
			header, msg := pgtlib.PopStr(msg)

			switch header {
			case pgtlib.C_CLIENT:
				broker.ClientMsg(sender, msg)
			case pgtlib.W_WORKER:
				broker.WorkerMsg(sender, msg)
			default:
				log.Printf("E: invalid message (sender=%s, header=%s): %q\n", sender, header, msg)
			}
		}
		//  Disconnect and delete any expired workers
		//  Send heartbeats to idle workers if needed

		if time.Now().After(broker.heartbeat_at) {
			broker.Purge()
			for _, worker := range broker.waiting {
				worker.Send(pgtlib.W_HEARTBEAT, []string{})
			}
			broker.heartbeat_at = time.Now().Add(HEARTBEAT_INTERVAL)
		}
	}
}