// closePorts closes all active ports and terminates ZMQ context
func closePorts() {
	log.Println("Closing ports...")
	optionsPort.Close()
	inPort.Close()
	outPort.Close()
	zmq.Term()
}

func closePorts() {
	optsPort.Close()
	tmplPort.Close()
	propPort.Close()
	if errPort != nil {
		errPort.Close()
	}
	zmq.Term()
}

// closePorts closes all active ports and terminates ZMQ context
func closePorts() {
	log.Println("Closing ports...")
	optionsPort.Close()
	inPort.Close()
	if errPort != nil {
		errPort.Close()
	}
	zmq.Term()
}

// closePorts closes all active ports and terminates ZMQ context
func closePorts() {
	log.Println("Closing ports...")
	optionsPort.Close()
	if inPort != nil {
		inPort.Close()
	}
	if outPort != nil {
		outPort.Close()
	}
	zmq.Term()
}

func closePorts() {
	requestPort.Close()
	failPort.Close()
	for _, p := range patternPorts {
		p.Close()
	}
	for _, p := range successPorts {
		p.Close()
	}
	zmq.Term()
}

// Close closes the server, usually deferred upon starting it
func (server *Server) Close() {
	// Don't do anything is already closing
	if server.closing {
		return
	}
	server.closing = true
	glog.Info("zrpc: server closing...")
	// Should block until all requests are done
	server.conn.Close()
	// Terminates the entire zmq context
	zmq.Term()
}

func (client *Client) Close() error {
	glog.Info("zrpc: client closing")
	client.mutex.Lock()
	if client.closing {
		client.mutex.Unlock()
		return ErrShutdown
	}
	client.closing = true
	client.mutex.Unlock()

	// These should block until all messages returned
	client.conn.Close()
	client.router.Close()

	// Terminate the zmq context
	return zmq.Term()
}

func main() {
	info := "greenline: notoriously unreliable\n" +
		"https://github.com/formwork-io/greenline\n" +
		"This is free software with ABSOLUTELY NO WARRANTY."
	fmt.Printf("%s\n--\n", info)
	var rails []Rail
	if len(os.Args) == 2 {
		var err error
		rails, err = ReadConfigFile(os.Args[1])
		if err != nil {
			die(err.Error())
		}
	} else {
		var err error
		rails, err = ReadEnvironment()
		if err != nil {
			die(err.Error())
		}
	}
	pprint("configuring %d rails", len(rails))

	socketPairs := make(map[*zmq.Socket]*zmq.Socket)
	socketNames := make(map[*zmq.Socket]string)
	poller := zmq.NewPoller()

	railmsg := "%s protocol %s, %d -> %d"

	for _, rail := range rails {

		var ingress *zmq.Socket
		var egress *zmq.Socket
		switch rail.Protocol {
		case "broadcast":
			ingress, egress = railToPubSub(&rail, poller)
		case "request":
			ingress, egress = railToRouterDealer(&rail, poller)
		default:
			die("The protocol %s is not valid.", rail.Protocol)
		}

		titledProtocol := strings.Title(rail.Protocol)
		pprint(railmsg, titledProtocol, rail.Name, rail.Ingress, rail.Egress)

		socketPairs[ingress] = egress
		socketNames[ingress] = fmt.Sprintf("%s (ingress)", rail.Name)

		socketPairs[egress] = ingress
		socketNames[egress] = fmt.Sprintf("%s (egress)", rail.Name)

		defer ingress.Close()
		defer egress.Close()
	}

	pprint("greenline alive")
	exitchan := make(chan os.Signal, 0)
	signal.Notify(exitchan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	go func() {
		sig := <-exitchan
		out("received %s signal, exiting.\n", sig.String())
		os.Exit(0)
	}()

	reloadchan := make(chan int)
	go reloader(reloadchan)
	readychan := make(chan bool)
	pollchan := make(chan bool)
	go func() {
		for {
			sockets, err := poller.Poll(-1)
			if err != nil {
				// get EINTR while polling?
				if IsEINTR(err) {
					// Continue polling; EINTR is normal for us...
					continue
					// ... with our use of signals.
				}

				// otherwise shutdown
				readychan <- false
				break
			}
			if len(sockets) != 0 {
				readychan <- true
			}
			// wait to poll once msgs processed
			<-pollchan
		}
	}()
	pprint("greenline ready")
	for {
		select {
		case reloadOp := <-reloadchan:
			if reloadOp&BinReload == BinReload {
				pprint("new binary available, restarting greenline")
				for key, value := range socketPairs {
					key.Close()
					value.Close()
				}
				zmq.Term()
				// exec or die
//.........这里部分代码省略.........

func TestLotsOfIO(t *testing.T) {
	const (
		numSockets = 503
	)

	defer zmq.Term()

	done := make(chan struct{})

	for i := 0; i < numSockets; i++ {
		addr := fmt.Sprintf("inproc://%d", i)

		numMessages := i * 11
		if i&1 == 1 {
			numMessages -= 1317
		}

		go func(addr string, numMessages int) {
			defer func() {
				done <- struct{}{}
			}()

			s, err := zmq.NewSocket(zmq.PULL)
			if err != nil {
				t.Fatal(err)
			}
			defer io.Remove(s)

			if err := s.Bind(addr); err != nil {
				t.Fatal(err)
			}

			c := make(chan zmqchan.Data)

			if err := io.Add(s, nil, c); err != nil {
				t.Fatal(err)
			}

			for n := 0; n < numMessages; n++ {
				m := <-c

				ms, err := strconv.Atoi(m.String())
				if err != nil {
					t.Fatal(err)
				}

				if ms != n {
					t.Fatalf("%d: %d != %d", i, ms, n)
				}
			}
		}(addr, numMessages)

		go func(addr string, numMessages int) {
			defer func() {
				done <- struct{}{}
			}()

			s, err := zmq.NewSocket(zmq.PUSH)
			if err != nil {
				t.Fatal(err)
			}
			defer io.Remove(s)

			if err := s.Connect(addr); err != nil {
				t.Fatal(err)
			}

			c := make(chan zmqchan.Data)
			defer close(c)

			if err := io.Add(s, c, nil); err != nil {
				t.Fatal(err)
			}

			for n := 0; n < numMessages; n++ {
				c <- zmqchan.Data{
					Bytes: []byte(strconv.Itoa(n)),
				}
			}
		}(addr, numMessages)
	}

	for i := 0; i < numSockets; i++ {
		<-done
	}
}