report error msg in each ReadLine
@@ -50,7 +50,8 @@ type Reader struct {
status int32
readChan chan Details
- errChan chan error
+ initErr error
+ initErrLock sync.RWMutex
currentPath string
address string
@@ -83,7 +84,7 @@ func NewReader(meta *reader.Meta, conf conf.MapConf) (reader.Reader, error) {
meta: meta,
status: StatusInit,
readChan: make(chan Details, len(paths)),
- errChan: make(chan error, 1),
+ initErrLock: sync.RWMutex{},
address: address,
paths: paths,
}, nil
@@ -125,7 +126,9 @@ func (r *Reader) Start() error {
go func() {
if err := r.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Errorf("Runner[%v] %q daemon start HTTP server failed: %v", r.meta.RunnerName, r.Name(), err)
- r.errChan <- err
+ r.initErrLock.Lock()
+ r.initErr = err
+ r.initErrLock.Unlock()
}
}()
log.Infof("Runner[%v] %q daemon has started", r.meta.RunnerName, r.Name())
@@ -149,16 +152,15 @@ func (r *Reader) ReadLine() (string, error) {
r.currentPath = data.Path
return data.Content, nil
case <-timer.C:
- select {
- case err := <-r.errChan:
- log.Errorf("Reader %s failed: %v", r.Name(), err)
- return "", err
- default:
- return "", nil
}
+
+ return "", r.FetchInitError()
}
- return "", nil
+func (r *Reader) FetchInitError() error {
+ r.initErrLock.RLock()
+ defer r.initErrLock.RUnlock()
+ return r.initErr
}
func (_ *Reader) SyncMeta() {}
