()
| 121 | } |
| 122 | |
| 123 | func (m *Source) Run() error { |
| 124 | defer m.Ctx.Recover() |
| 125 | defer close(m.msgOutCh) |
| 126 | |
| 127 | if m.Scanner == nil { |
| 128 | u.Warnf("no datasource configured?") |
| 129 | return fmt.Errorf("No datasource found") |
| 130 | } |
| 131 | |
| 132 | sigChan := m.SigChan() |
| 133 | |
| 134 | for item := m.Scanner.Next(); item != nil; item = m.Scanner.Next() { |
| 135 | |
| 136 | select { |
| 137 | case <-sigChan: |
| 138 | return nil |
| 139 | case m.msgOutCh <- item: |
| 140 | // continue |
| 141 | } |
| 142 | |
| 143 | } |
| 144 | return nil |
| 145 | } |