// generateRandomBytes returns securely generated random bytes. It will return
// an error if the system's secure random number generator fails to function
// correctly, in which case the caller should not continue.
func generateRandomBytes(n int) []byte {
	b := make([]byte, n)
	_, err := rand.Read(b)
	if err != nil {
		lg.Warningln("could not generate secure random key, using non secure random instead")
		lg.Warningln(err)
		for i := 0; i < n; i++ {
			b[i] = byte(mrand.Intn(256))
		}
	}
	return b
}

func startAnalysis(clients db.Clients) {
	go func() {
		tick := time.NewTicker(10 * time.Second)

		lastID, err := clients.DB.GetLastProcessedSampleID()
		if err != nil {
			lg.Warningln(err)
		}
		lg.Infof("starting analysis from sample ID %d", lastID)

		lastPersistedID := lastID

		for range tick.C {
			results, err := clients.DB.GetSamples(uint64(lastID), "")
			if err != nil {
				lg.Fatal(err)
			}
			n := 0
			start := time.Now()
		loop:
			for s := range results {
				n++
				if s.ID > lastID {
					lastID = s.ID
				}
				if s.Type == "NewClientToken" {
					if !shared.AcceptedHost(s.Host) {
						lg.Warningln("not accepted host id:", s.ID, s.Host)
						continue loop
					}
					err := clients.DB.PublishHost(s)
					if err != nil {
						lg.Warning(err)
					}
				}
			}
			if n != 0 && lg.V(15) {
				lg.Infof("processed %d samples in %s", n, time.Since(start).String())
			}
			if lastID != lastPersistedID {
				err = clients.DB.SetLastProcessedSampleID(lastID)
				if err != nil {
					lg.Errorln(err)
				} else {
					lastPersistedID = lastID
				}
			}
		}
	}()
}

// errToAPIStatus converts an error to an api.Status object.
func ErrToAPIStatus(err error) *api.Status {
	switch t := err.(type) {
	case statusError:
		status := t.Status()
		if len(status.Status) == 0 {
			status.Status = api.StatusFailure
		}
		if status.Code == 0 {
			switch status.Status {
			case api.StatusSuccess:
				status.Code = http.StatusOK
			case api.StatusFailure:
				status.Code = http.StatusInternalServerError
			}
		}
		//TODO: check for invalid responses
		return &status
	default:
		status := http.StatusInternalServerError

		// Log errors that were not converted to an error status
		// by REST storage - these typically indicate programmer
		// error by not using pkg/api/errors, or unexpected failure
		// cases.
		lg.Warningln(fmt.Errorf("apiserver received an error that is not an api.Status: %v", err))

		return &api.Status{
			Status:  api.StatusFailure,
			Code:    status,
			Reason:  api.StatusReasonUnknown,
			Message: err.Error(),
		}
	}
}

func (s *singleUseAuthKeyStore) New() string {

	b := make([]byte, 10)
	_, err := rand.Read(b)
	if err != nil {
		lg.Warningln("could not generate secure random key, using non secure random instead")
		lg.Warningln(err)
		for i := 0; i < len(b); i++ {
			b[i] = byte(mrand.Intn(256))
		}
	}
	key := hex.EncodeToString(b)

	s.Lock()
	defer s.Unlock()
	s.entries[key] = time.Now()
	return key
}

func (d *DB) SetLastProcessedSampleID(id uint64) error {
	psql := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar)
	idstr := strconv.FormatUint(id, 10)

	s := psql.
		Update("central_state").
		Set("value", idstr).
		Where(squirrel.Eq{"name": "last_processed_sample_id"})
	sr, err := s.RunWith(d.cache).Exec()
	if err != nil {
		logSQLErr(err, &s)
		log.Fatal(err)
	}
	sn, err := sr.RowsAffected()
	if err != nil {
		lg.Warningln(err)
		return err
	}
	if sn > 0 {
		return nil
	}

	i := psql.
		Insert("central_state").
		Columns("name", "value").
		Values("last_processed_sample_id", idstr)

	ir, err := i.RunWith(d.cache).Exec()
	if err != nil {
		logSQLErr(err, &i)
		log.Fatal(err)
	}
	in, err := ir.RowsAffected()
	if err != nil {
		lg.Warningln(err)
		return err
	}
	if in > 0 {
		return nil
	}
	log.Fatal("unknown error")
	return nil
}

// Wait blocks until the underlying process is stopped
func (s *Service) Wait() {
	if s.cmd != nil {
		lg.V(10).Infof("Waiting for process %s to exit", s.ID)
		err := s.cmd.Wait()
		if err != nil {
			lg.Warningln(err)
		}
		lg.V(10).Infof("%s exited", s.ID)
	}
	s.waiter.Wait()

}

// AtexitKillCmd takes care of killing a command on application exit.
//
// TODO: currently this does not clean up references to dead processes, it just
// adds forever.
func AtexitKillCmd(cmd *exec.Cmd) {
	Atexit(func() {
		lg.V(10).Info("Atexit kill ", cmd.Path, cmd.Args)
		err := cmd.Process.Kill()
		if err != nil {
			lg.V(5).Info("kill failed:", err)
		}
		// TODO: possible deadlock?
		if err := cmd.Wait(); err != nil {
			lg.Warningln(err)
		}
	})
}

func StartAnalysis(clients db.Clients) {

	tick := time.NewTicker(5 * time.Second)
	lastID, err := clients.DB.GetLastProcessedSampleID()
	if err != nil {
		lg.Warningln(err)
	}

	for n := 0; n < 4; n++ {
		go sessionFetcher(clients)
		go samplesAnalyzer()
		go hostPublisher(clients)
	}

	lg.Infof("starting analysis from sample ID %d", lastID)

	lastPersistedID := lastID

	for range tick.C {
		results, err := clients.DB.GetSamples(uint64(lastID), "")
		if err != nil {
			lg.Errorf("database err (skipping): %v", err)
			continue
		}
		n := 0
		start := time.Now()
		for s := range results {
			n++
			if s.ID > lastID {
				lastID = s.ID
			}

			if s.Origin == "Central" && s.Type == "HTTPHeader" {
				sessionFetchC <- s.Token
			}
		}
		if n != 0 && lg.V(15) {
			lg.Infof("processed %d samples in %s", n, time.Since(start).String())
		}
		if lastID != lastPersistedID {
			err = clients.DB.SetLastProcessedSampleID(lastID)
			if err != nil {
				lg.Errorln(err)
			} else {
				lastPersistedID = lastID
			}
		}
	}

}

// NewUpdateChecker creates and returns an UpdateChecker instance.
// The caller should then listen on the RequestC channel for UpdateRequests.
func NewUpdateChecker(name string) (*UpdateChecker, error) {

	c := &UpdateChecker{
		Interval: time.Duration(1*time.Hour + (time.Minute * (time.Duration(rand.Intn(120))))),
	}
	c.response = make(chan UpdateResult)
	c.RequestC = make(chan UpdateRequest)
	c.forceRequestC = make(chan bool)

	lg.Infof("Setting up update timer for %s every %f minute(s) ",
		name, c.Interval.Minutes())
	ticker := time.NewTicker(c.Interval)
	go func() {
		for {
			select {
			case <-c.forceRequestC:
				if !c.active {
					continue
				}
				c.RequestC <- UpdateRequest{
					ResponseC: c.response,
				}
			case <-ticker.C:
				if !c.active {
					continue
				}
				c.RequestC <- UpdateRequest{
					ResponseC: c.response,
				}
			case response := <-c.response:
				c.LastCheck = time.Now()
				switch response {
				case UpdateSuccess:
					lg.V(5).Infoln("UpdateSuccess")
					c.LastUpdate = c.LastCheck
					c.LastFailedCheck = time.Time{}
				case UpdateError:
					lg.Warningln("update check failed")
					c.LastFailedCheck = c.LastCheck
					<-time.After(3*time.Second + time.Duration(rand.Intn(5)))
					go func() {
						c.forceRequestC <- true
					}()
				}
			}

		}
	}()
	return c, nil
}

// Wait blocks until the underlying process is stopped
func (s *Service) wait() {
	if s.isCopy {
		lg.Fatal("wait called on copy of service!")
	}
	if s.cmd != nil {
		lg.V(10).Infof("Waiting for process %s to exit", s.ID)
		err := s.cmd.Wait()
		if err != nil {
			lg.Warningln(err)
		}
		lg.V(10).Infof("%s exited", s.ID)
	}
	s.waiter.Wait()

}

func connect(connection shared.Connection, authKey string) {

	defaultTransportM.RLock()
	if defaultTransport != nil {
		err := defaultTransport.Remove()
		if err != nil {
			lg.Warningln(err)
		}
	}
	defaultTransportM.RUnlock()

	event := newConnectionEventhistory(connection)

	event.newState(ServiceInit)
	ts, err := NewTransportService(connection)
	ts.authSecret = authKey
	if lg.V(6) {
		ts.SetVerbose()
	}
	ts.SetBindaddr(DefaultProxyBindAddr)

	if err != nil {
		event.newState(Failed)
		event.newState(Ended)
		return
	}
	event.newState(ServiceStart)
	event.ServiceID = ts.Service.ID
	err = ts.Start()
	if err != nil {
		event.newState(Failed)
		event.newState(Ended)
		return
	}
	response := ts.Service.Response
	if response["protocol"] != "socks5" {
		event.newState(WrongProtocol)
		event.newState(Failed)
	}

	event.newState(Test)
	go testConn(&event)

	defaultTransportM.Lock()
	defaultTransport = ts
	defaultTransportM.Unlock()

}

// NewDebugResponse creates a filled DebugResponse struct
func NewDebugResposne(version string, config interface{}) *DebugResponse {
	ID, err := shared.SecureRandomString(12)
	if err != nil {
		panic("could not generate random number")
	}
	response := &DebugResponse{
		Header: DebugHeader{
			Cmd:       filepath.Base(os.Args[0]),
			ID:        ID,
			Version:   version,
			CreatedAt: time.Now(),
			OS:        runtime.GOOS,
			Arch:      runtime.GOARCH,
			GoVersion: runtime.Version(),
		},
		Config: config,
	}

	getProfile := func(name string) []string {
		buf := bytes.NewBuffer(nil)
		err := pprof.Lookup(name).WriteTo(buf, 2)
		if err != nil {
			lg.Errorln(err)
		} else {
			return strings.Split(
				buf.String(),
				"\n")
		}
		return []string{}
	}

	response.Heap = getProfile("heap")
	response.GoRoutines = getProfile("goroutine")
	response.ThreadCreate = getProfile("threadcreate")
	response.Block = getProfile("block")
	// memlog should be last so that it can catch errors up to the point of
	// collection.
	response.Log = lg.Memlog()

	if PublicKey != "" {
		response.Encrypt(PublicKey)
	} else {
		lg.Warningln("PublicKey not set, exporting unencrypted debug log")
	}

	return response

}

// GetSuggestions returns all local/remote suggestion sessions, can be filtered
// by time of creation.
func GetSuggestions(w rest.ResponseWriter, r *rest.Request) {
	var res []client.Suggestion
	all := client.AllSuggestions()
	after := r.URL.Query().Get("after")

	if after == "" {
		res = all
	} else {
		after, err := time.Parse(time.RFC3339, after)
		if err != nil {
			lg.Warningln("Wrong time format: %s", after)
			return
		}
		for _, v := range all {
			if v.CreatedAt.After(after) {
				res = append(res, v)
			}
		}
	}
	w.WriteJson(res)
}

// StartBinaryUpgradeChecker checks for binary upgrades when the connection is
// up and on a schedule.
//
// This function runs in it's own goroutine.
func StartBinaryUpgradeChecker(diffsBaseURL string) {
	if !upgradeEnabled {
		lg.Infoln("binary upgrades are disabled using the command line flag")
		return
	}
	if VERSION == "" {
		lg.Warningln("VERSION not set, binary upgrades are disabled")
		return
	}
	_, err := version.NewVersion(VERSION)
	if err != nil {
		lg.Warningf("VERSION '%s' is not a valid semver version, binary upgrades are disabled: %v", VERSION, err)
		return
	}

	connectionEventListener := make(chan service.ConnectionHistory)
	uChecker, _ := NewUpdateChecker("binary")
	service.AddListener(connectionEventListener)
	for {
		select {
		// Update when the transport connection comes up
		case event := <-connectionEventListener:
			if event.IsUp() {
				uChecker.Activate()
				uChecker.UpdateNow()
			}

		// Update by request of the update checker
		case request := <-uChecker.RequestC:
			err := upgradeBinaryCheck(diffsBaseURL)
			if err != nil {
				lg.Errorln(err)
				request.ResponseC <- UpdateError
			} else {
				request.ResponseC <- UpdateSuccess
			}
		}
	}
}

func StartConnectionManager(authKey string) {
	go updateTransportOkLoop()

	listeners := make([]chan ConnectionHistory, 0)

	// TODO: Test on irregular intervals
	reverifyTicker := time.NewTicker(connectionManagerTimings.TestTransportTicker)

	// the key is Connection.UUID
	histories := make(map[string][]ConnectionEvent)
	currents := make(map[string]*ConnectionEvent)

	currentConnectionID := ""
	currentConnectionsMu.Lock()
	currentConnIdx := 0
	if len(currents) < 1 {
		go connect(currentConnections[currentConnIdx], authKey)
		currentConnectionID = currentConnections[currentConnIdx].ID
	}
	currentConnectionsMu.Unlock()

	firstUpNoProblems := true // no need to spam the user with popups

	var reconnectTimer *time.Timer
	var connectionTestRunning bool
loop:

	for {
	s:
		select {
		case <-stopCh:
			lg.Infoln("connection manager shut down")
			break loop

		case <-reconnectCh:
			currentConnectionsMu.Lock()
			if len(currentConnections) < 1 {
				currentConnectionsMu.Unlock()
				lg.Warningln("No connections enabled")
				reconnectTimer = time.AfterFunc(connectionManagerTimings.ReconnectTransportDelay, func() {
					reconnectCh <- true
				})
				break s
			}
			currentConnIdx = (currentConnIdx + 1) % (len(currentConnections))
			c := currentConnections[currentConnIdx]
			lg.V(10).Infof("reconnecting to transport %v", c)
			go connect(currentConnections[currentConnIdx], authKey)
			currentConnectionID = c.ID
			currentConnectionsMu.Unlock()

		case listener := <-addNetworkStateListener:
			listeners = append(listeners, listener)

		case event := <-connectionEvents:
			if _, v := histories[event.Connection.ID]; !v {
				histories[event.Connection.ID] = make([]ConnectionEvent, 0)
			} else if len(histories[event.Connection.ID]) > 20 {
				lg.V(5).Infoln("trimming connection history")
				histories[event.Connection.ID] = histories[event.Connection.ID][:20]
			}
			histories[event.Connection.ID] = append(histories[event.Connection.ID], event)
			currents[event.Connection.ID] = &event
			emitEvent := ConnectionHistory{
				History: histories[event.Connection.ID],
			}

			if lg.V(3) {
				switch event.State {
				case Failed, TestFailed:
					lg.Warningln("event  ", event.Connection.ID, ": ", event.State)
				default:
					lg.Infoln("event  ", event.Connection.ID, ": ", event.State)
				}
			}
			switch event.State {
			case Up:
				if firstUpNoProblems {
					firstUpNoProblems = false
				} else {
					ui.Notify("transport_connected_message")
				}
			case Failed:
				firstUpNoProblems = false
				ui.Notify("transport_error_message")
			case TestFailed:
				firstUpNoProblems = false
				ui.Notify("transport_retry")
			case Ended:
				delete(currents, event.Connection.ID)
				lg.V(15).Infoln("waiting before sending reconnect")
				if reconnectTimer != nil {
					reconnectTimer.Stop()
				}
				reconnectTimer = time.AfterFunc(connectionManagerTimings.ReconnectTransportDelay, func() {
					reconnectCh <- true
				})
			}
			lg.V(7).Infoln("Forwarding connection event to listeners", emitEvent.Current())
			for _, l := range listeners {
//.........这里部分代码省略.........

func samplesAnalyzer() {
loop:
	for samples := range sampleAnalysisC {
		var (
			newTokenSample db.Sample
			clientSamples  = make(map[string]db.Sample, 0)
			centralSamples = make(map[string]db.Sample, 0)
		)

		// organize input data
		for _, s := range samples {
			switch s.Type {
			case "NewClientToken":
				if newTokenSample.Token != "" {
					lg.Error("got more than one newTokenSample, aborting")
					continue loop
				}
				newTokenSample = s
			case "HTTPHeader", "DNSQuery":
				switch s.Origin {
				case "Central":
					centralSamples[s.Type] = s
				case "Client":
					clientSamples[s.Type] = s
				}
			default:
				lg.Errorf("dont know how to handle %d %s, skipping", s.ID, s.Type)
			}
		}

		// validate that wanted data types are available
		if newTokenSample.Token == "" {
			lg.Errorln("No newTokenSample, aborting")
			continue loop
		}
		if !shared.AcceptedHost(newTokenSample.Host) {
			lg.Warningln("not accepted host id:", newTokenSample.ID, newTokenSample.Host)
			continue loop
		}
		for _, smap := range []map[string]db.Sample{clientSamples, centralSamples} {
			for _, stype := range []string{"HTTPHeader"} {
				if _, ok := smap[stype]; !ok {
					lg.Errorf("missing %s, cannot analyse", stype)
					continue loop
				}
			}
		}

		// parse data
		clientSample := clientSamples["HTTPHeader"]
		var clientHeader measure.HTTPHeaderResult
		centralSample := centralSamples["HTTPHeader"]
		var centralHeader measure.HTTPHeaderResult

		if err := json.Unmarshal(clientSample.Data, &clientHeader); err != nil {
			lg.Errorln(err)
			continue loop
		}
		if err := json.Unmarshal(centralSample.Data, &centralHeader); err != nil {
			lg.Error(err)
			continue loop
		}

		// score data
		score := scoreHTTPHeaders(clientHeader, centralHeader)
		lg.V(10).Infof("session:%s %s", newTokenSample.Token, score)

		if score.Score() >= 0.5 {
			lg.Infoln("publishing session", newTokenSample.Token)
			hostPublishC <- newTokenSample
		} else {
			lg.Infoln("not publishing session", newTokenSample.Token)
		}
	}
}

// SubmitSuggestion initiates the comminication with Central for a Submission
// session.
func SubmitSuggestion(w rest.ResponseWriter, r *rest.Request) {

	// // TODO This is the response that must be forwarded from central/api and parsed by client and passed on to the browser.
	// apiutils.WriteRestError(w,
	// 	apierrors.NewInvalid("object", "suggestion",
	// 		fielderrors.ValidationErrorList{
	// 			fielderrors.NewFieldValueNotSupported("URL", "...", []string{})}))
	// return

	ID := r.PathParam("id")
	suggestion, ok := client.GetSuggestion(ID)
	if !ok {
		apiutils.WriteRestError(w, apierrors.NewNotFound("suggestion", ID))
		return
	}
	wanip := shared.GetPublicIPAddr()
	if wanip == nil {
		lg.Warning("could not resolve public ip addr")
	}

	conf := clientconfig.Get()
	restclient, err := NewRestClient()
	if err != nil {
		apiutils.WriteRestError(w, apierrors.NewInternalError(err))
		return
	}

	tokenResp, err := suggestion.RequestToken(
		restclient, wanip, conf.Settings.Local.CountryCode)
	if err != nil {
		if apiutils.IsNetError(err) {
			apiutils.WriteRestError(w, apierrors.NewServerTimeout("alkasir-central", "request-submission-token", 0))
		} else {
			apiutils.WriteRestError(w, apierrors.NewInternalError(err))
		}
		return
	}
	n, err := suggestion.SendSamples(restclient)
	if err != nil {
		lg.Warningln("error sending samples", err.Error())
	}
	lg.V(5).Infof("sent %d samples", n)

	// continue sending samples if future measuremetns are expected to come
	prepared, err := suggestion.Prepared()
	if err != nil {
		lg.Errorln(err)
	} else if !prepared {
		lg.V(5).Infof("all samples not collected, will try to send the rest when they are done")
		go func(s client.Suggestion) {
			start := time.Now()
			t := time.NewTicker(30 * time.Second)
			defer t.Stop()

		submitSamples:
			for range t.C {
				if time.Now().After(start.Add(15 * time.Minute)) {
					lg.Errorln("Stopping trying to send additional samples")
					return
				}

				prepared, err := suggestion.Prepared()
				if err != nil {
					lg.Errorln(err)
					return
				}
				if prepared {
					restclient, err := NewRestClient()
					if err != nil {
						continue submitSamples
					}
					n, err := suggestion.SendSamples(restclient)
					lg.V(5).Infof("sent %d samples", n)
					if err != nil {
						lg.Warningln("error sending samples", err.Error())
						continue submitSamples
					}
					return
				}
			}
		}(suggestion)
	}

	u, err := url.Parse(suggestion.URL)
	if err != nil {
		lg.Errorln(err)
	} else {
		err := clientconfig.Update(func(conf *clientconfig.Config) error {
			conf.BlockedHosts.Add(u.Host)
			lastBlocklistChange = time.Now()

			pac.UpdateBlockedList(conf.BlockedHostsCentral.Hosts,
				conf.BlockedHosts.Hosts)
			return nil
		})
		if err != nil {
			lg.Errorln(err)
		}
//.........这里部分代码省略.........

// Initialize service
func (s *Service) initService() error {
	cmd := s.cmd
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	s.stdout = stdout
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return err
	}
	s.stderr = stderr

	stdin, err := cmd.StdinPipe()
	if err != nil {
		return err
	}
	s.stdin = stdin

	if lg.V(5) {
		alkasirEnv := ""
		for _, v := range s.cmd.Env {
			if strings.HasPrefix(v, "ALKASIR_") {
				alkasirEnv += v + " "
			}
		}
		lg.Infof("Starting service: %s %s", alkasirEnv, cmd.Path)
	}
	err = cmd.Start()
	if err != nil {
		return err
	}

	scanner := bufio.NewScanner(stdout)
	var transportErrMsg string
	transportErr := false
	var line string
	for scanner.Scan() {
		line = scanner.Text()

		lg.V(5).Infoln("DBG: ", line)

		if errorM.MatchString(line) {
			transportErr = true
			transportErrMsg = line
			return errors.New("error: " + transportErrMsg)
		} else if doneM.MatchString(line) {
			break
		} else if exposeM.MatchString(line) {
			match := exposeM.FindStringSubmatch(line)
			s.Response["bindaddr"] = match[3]
			s.Response["protocol"] = match[2]
			s.registerMethod(match[1], match[2], match[3])
		} else if versionM.MatchString(line) {
		} else if parentM.MatchString(line) {
			match := parentM.FindStringSubmatch(line)
			s.Response["parentaddr"] = match[3]
		} else {
			lg.Infoln("not handled line:", line)
			return errors.New("unhandeled line")
		}
	}
	if transportErr {
		err := cmd.Wait()
		if err != nil {
			lg.Warningln(err)
		}
		lg.Fatal(transportErrMsg)
		return errors.New("transport err")
	}
	return err
}

// Run runs the initialized server.
func Run() {
	var wg sync.WaitGroup

	// start monitor server
	go startMonitoring(*monitorBindAddr)

	// start the getpublic ip updater.
	go func() {
		_ = shared.GetPublicIPAddr()
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		lg.V(2).Infoln("Loading recent sessions from postgres...")
		recents, err := sqlDB.RecentSuggestionSessions(20000)
		if err != nil {
			lg.Fatal(err)
		}
		db.SessionTokens.Reset(recents)
		lg.V(2).Infof("Loaded %d sessions from postgres...", len(recents))
		lg.Flush()

	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		conn := redisPool.Get()
		defer conn.Close()
		lg.V(2).Infoln("BGPDump refresh started...")
		n, err := internet.RefreshBGPDump(conn)
		lg.V(2).Infof("BGPDump refresh ended, %d items added.", n)
		lg.Flush()
		if err != nil {
			if *offline {
				lg.Infoln("offline", err)
			} else {
				lg.Fatal(err)
			}
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		conn := redisPool.Get()
		defer conn.Close()
		lg.V(2).Infoln("CIDRReport refresh started...")
		n, err := internet.RefreshCIDRReport(conn)
		lg.V(2).Infof("CIDRReport refresh ended, %d items added", n)
		if err != nil {
			if *offline {
				lg.Infoln("offline", err)
			} else {
				lg.Fatal(err)
			}
		}
	}()
	wg.Wait()

	// start signal handling
	wg.Add(1)
	go func() {
		ch := make(chan os.Signal)
		signal.Notify(ch, syscall.SIGINT)
		lg.Infoln(<-ch)
		wg.Done()
	}()

	internetClient := db.NewInternetClient(redisPool)
	maxmindClient := db.NewMaxmindClient(mmCountryDB, mmCityDB)

	clients := db.Clients{
		DB:       sqlDB,
		Internet: internetClient,
		Maxmind:  maxmindClient,
	}

	// start http json api server
	go func(addr string, dba db.Clients) {
		mux, err := apiMux(dba)
		lg.Info("Starting http server", addr)
		err = http.ListenAndServe(addr, mux)
		if err != nil {
			lg.Fatal(err)
		}
	}(*apiBindAddr, clients)

	// start http export api server
	go func(addr string, dba db.Clients) {
		if *exportApiSecretKey == "" {
			lg.Warningln("exportApiSecretKey flag/env not set, will not start export api server")
			b := make([]byte, 32)
			_, err := rand.Read(b)
			if err != nil {
				lg.Fatalf("random generator not functioning...")
				return
			}
//.........这里部分代码省略.........

// initDone means handing off the service process output to it's own goroutine.
func (s *Service) initDone() error {

	lg.V(5).Infof("s.request: %+v", s.Request)
	lg.V(5).Infof("s.response: %+v", s.Response)
	s.waiter.Add(1)
	s.running = true
	go func() {
		defer func() {
			s.running = false
		}()

		go func() {
			scanner := bufio.NewScanner(s.stdout)
			for scanner.Scan() {
				if scanner.Err() != nil {
					lg.Warningln("service stdout", scanner.Err())
					break
				}
				line := scanner.Text()
				lg.Infoln(s.ID, "stdout:", line)
			}
			lg.V(20).Infof("service outch closed %s", s.ID)
		}()

		go func() {
			scanner := bufio.NewScanner(s.stderr)
			for scanner.Scan() {
				if scanner.Err() != nil {
					lg.Warningln("service stderr", scanner.Err())
					break
				}
				line := scanner.Text()
				lg.Infoln(s.ID, "stderr:", line)
			}
			lg.V(20).Infof("service stderr closed %s", s.ID)
		}()

		s.quit = make(chan bool)
		defer close(s.quit)

		select {
		case <-s.quit:
			lg.V(6).Infof("stopping service %s", s.ID)

			if err := s.stdin.Close(); err != nil {
				lg.Errorf("could not close stdin for %s: %s", s.ID, err.Error())
			}

			if err := s.stdout.Close(); err != nil {
				lg.Errorf("could not close stdout for %s: %s", s.ID, err.Error())
			}

			if err := s.stderr.Close(); err != nil {
				lg.Errorf("could not close stderr for %s: %s", s.ID, err.Error())
			}

			lg.V(10).Infof("Killing process service %s", s.ID)

		}

		lg.V(10).Infof("stopped service %s", s.ID)
		if s.removeOnStop {
			ManagedServices.remove(s)
		}
		s.waiter.Done()
	}()
	return nil
}