| 124 | } |
| 125 | |
| 126 | func (u *S3) Run(upch <-chan string) error { |
| 127 | // Stop blocks until the upload goroutine has exited. |
| 128 | defer u.Stop() |
| 129 | |
| 130 | // Use a buffered channel to allow an extra message to be pushed by |
| 131 | // the deferred function in the goroutine when the Run function |
| 132 | // exits because of an error from u.uploadDirectory. |
| 133 | // An unbuffered channel will cause a deadlock because u.wgUpload.Done() |
| 134 | // is never reached |
| 135 | errCh := make(chan error, 1) |
| 136 | |
| 137 | // Start a goroutine in which we periodically look at the source |
| 138 | // path for files and upload the ones we find. |
| 139 | u.wgUpload.Add(1) |
| 140 | go func() { |
| 141 | ticker := time.NewTicker(u.Cfg.Interval) |
| 142 | defer func() { |
| 143 | ticker.Stop() |
| 144 | log.Info("starting last upload") |
| 145 | if err := u.uploadDirectory(); err != nil { |
| 146 | log.Errorf("can't complete last upload: %v", err) |
| 147 | } |
| 148 | log.Info("completed last upload") |
| 149 | u.wgUpload.Done() |
| 150 | }() |
| 151 | |
| 152 | for { |
| 153 | select { |
| 154 | case <-ticker.C: |
| 155 | if err := u.uploadDirectory(); err != nil { |
| 156 | if u.Cfg.ExitOnError { |
| 157 | errCh <- err |
| 158 | return |
| 159 | } |
| 160 | log.Error(err) |
| 161 | } |
| 162 | case <-u.quit: |
| 163 | return |
| 164 | } |
| 165 | } |
| 166 | }() |
| 167 | |
| 168 | for { |
| 169 | select { |
| 170 | case err := <-errCh: |
| 171 | return err |
| 172 | case sourceFilePath, more := <-upch: |
| 173 | if !more { |
| 174 | return nil |
| 175 | } |
| 176 | err := u.move(sourceFilePath) |
| 177 | atomic.AddInt64(&u.totaln, int64(1)) |
| 178 | atomic.AddInt64(&u.queuedn, int64(1)) |
| 179 | if err != nil { |
| 180 | if u.Cfg.ExitOnError { |
| 181 | return fmt.Errorf("couldn't move: %v", err) |
| 182 | } |
| 183 | log.WithFields(log.Fields{"filepath": sourceFilePath}).WithError(err).Error("couldn't move") |