func (h *HostCursors) Update(id string, other *utils.HostCursor) {
	h.mtx.Lock()
	defer h.mtx.Unlock()
	curr, ok := h.Data[id]
	if !ok || other.After(*curr) {
		h.Data[id] = other
	}
}

func (m *Mux) addAggregator(addr string) {
	l := m.logger.New("fn", "addAggregator", "addr", addr)
	// TODO(titanous): add dial timeout
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		l.Error("failed to connect to aggregator", "error", err)
		return
	}
	l.Info("connected to aggregator")

	host, _, _ := net.SplitHostPort(addr)
	c, _ := client.New("http://" + host)
	cursors, err := c.GetCursors()
	if err != nil {
		// TODO(titanous): retry
		l.Error("failed to get cursors from aggregator", "error", err)
		conn.Close()
		return
	}

	var aggCursor *utils.HostCursor
	if c, ok := cursors[m.hostID]; ok {
		aggCursor = &c
	}
	if aggCursor != nil {
		l.Info("got cursor", "cursor.timestamp", aggCursor.Time, "cursor.seq", aggCursor.Seq)
	} else {
		l.Info("no cursor for host")
	}

	appLogs, err := m.logFiles("")
	if err != nil {
		l.Error("failed to get local log files", "error", err)
		conn.Close()
		return
	}

	bufferedMessages := make(chan message)
	firehose := make(chan message)
	done := make(chan struct{})

	// subscribe to all messages
	unsubscribe := m.subscribe(firehoseApp, firehose)

	bufferCursors := make(map[string]utils.HostCursor)
	var bufferCursorsMtx sync.Mutex
	go func() {
		l := m.logger.New("fn", "sendToAggregator", "addr", addr)
		defer unsubscribe()
		defer conn.Close()
		defer close(done)
		bm := bufferedMessages // make a copy so we can nil it later
		for {
			var m message
			var ok bool
			select {
			case m, ok = <-bm:
				if !ok {
					bm = nil
					continue
				}
			case m, ok = <-firehose:
				if !ok {
					return
				}

				// if app in list of app logs and cursor from reading files, skip
				appID := string(m.Message.AppName)
				if _, ok := appLogs[appID]; ok {
					bufferCursorsMtx.Lock()
					c, ok := bufferCursors[appID]
					bufferCursorsMtx.Unlock()
					if !ok || c.After(*m.HostCursor) {
						continue
					}
				}
			}
			if _, err := conn.Write(rfc6587.Bytes(m.Message)); err != nil {
				l.Error("failed to write message", "error", err)
				return
			}
		}
	}()

	for appID, logs := range appLogs {
		for i, name := range logs {
			func() {
				l := l.New("log", name)
				f, err := os.Open(name)
				if err != nil {
					l.Error("failed to open log file", "error", err)
					return
				}
				defer f.Close()
				sc := bufio.NewScanner(f)
				sc.Split(rfc6587.SplitWithNewlines)
				var cursor *utils.HostCursor
				cursorSaved := false
			scan:
				for sc.Scan() {
//.........这里部分代码省略.........