code: func (m *Mux) Follow(r io.ReadCloser, buffer string, msgID logagg.MsgID, config *Config) *LogStream {
  	hdr := &rfc5424.Header{
  		Hostname: []byte(config.HostID),
  		AppName:  []byte(config.AppID),
  		MsgID:    []byte(msgID),
  	}
  	if config.JobType != "" {
  		hdr.ProcID = []byte(config.JobType + "." + config.JobID)
  	} else {
  		hdr.ProcID = []byte(config.JobID)
  	}
  
  	s := &LogStream{
  		m:    m,
  		log:  r,
  		done: make(chan struct{}),
  	}
  	s.closed.Store(false)
  
  	m.jobsMtx.Lock()
  	defer m.jobsMtx.Unlock()
  	// set up the WaitGroup so that subscribers can track when the job fds have closed
  	wg, ok := m.jobWaits[config.JobID]
  	if !ok {
  		wg = &sync.WaitGroup{}
  		m.jobWaits[config.JobID] = wg
  	}
  	wg.Add(1)
  	if !ok {
  		// we created the wg, so create a goroutine to clean up
  		go func() {
  			wg.Wait()
  			m.jobsMtx.Lock()
  			defer m.jobsMtx.Unlock()
  			delete(m.jobWaits, config.JobID)
  		}()
  	}
  
  	// if there is a jobStart channel, a subscriber is waiting for the WaitGroup
  	// to be created, signal it.
  	if ch, ok := m.jobStarts[config.JobID]; ok {
  		close(ch)
  		delete(m.jobStarts, config.JobID)
  	}
  
  	go s.follow(r, buffer, config.AppID, hdr, wg)
  	return s
  }
docs: // Follow starts a goroutine that reads log lines from the reader into the mux.
  // It runs until the reader is closed or an error occurs. If an error occurs,
  // the reader may still be open.