| 116 | func (m *TaskSequential) Children() []Task { return m.tasks } |
| 117 | |
| 118 | func (m *TaskSequential) Run() (err error) { |
| 119 | defer m.Ctx.Recover() // Our context can recover panics, save error msg |
| 120 | defer func() { |
| 121 | //close(m.msgOutCh) // closing output channels is the signal to stop |
| 122 | //u.Debugf("close TaskSequential: %v", m.Type()) |
| 123 | }() |
| 124 | |
| 125 | var wg sync.WaitGroup |
| 126 | |
| 127 | // Either of the SigQuit, or error channel will |
| 128 | // cause breaking out of task execution below |
| 129 | // go func() { |
| 130 | // select { |
| 131 | // case err := <-m.errCh: |
| 132 | // u.Errorf("error on run %v", err) |
| 133 | // case <-m.sigCh: |
| 134 | // u.Warnf("%p %q got quit channel?", m, m.Name) |
| 135 | // // If we close here, we close without draining not giving messaging time |
| 136 | // // so we should???? |
| 137 | // //err = m.Close() |
| 138 | // // for _, task := range m.runners { |
| 139 | // // task.Quit() |
| 140 | // // } |
| 141 | // } |
| 142 | // }() |
| 143 | |
| 144 | // start tasks in reverse order, so that by time |
| 145 | // source starts up all downstreams have started |
| 146 | for i := len(m.runners) - 1; i >= 0; i-- { |
| 147 | wg.Add(1) |
| 148 | go func(taskId int) { |
| 149 | task := m.runners[taskId] |
| 150 | //u.Infof("starting task %d-%d %T in:%p out:%p", m.depth, taskId, task, task.MessageIn(), task.MessageOut()) |
| 151 | if taskErr := task.Run(); taskErr != nil { |
| 152 | u.Errorf("%T.Run() errored %v", task, taskErr) |
| 153 | // TODO: what do we do with this error? send to error channel? |
| 154 | err = taskErr |
| 155 | m.errors = append(m.errors, taskErr) |
| 156 | } |
| 157 | //u.Debugf("%p %q exiting taskId: %p %v %T", m, m.Name, task, taskId, task) |
| 158 | wg.Done() |
| 159 | // Lets look for the last task to shutdown, the result-writer or projection |
| 160 | // will finish first on limit so we need to shutdown sources |
| 161 | if len(m.runners)-1 == taskId { |
| 162 | //u.Warnf("%p got shutdown on last one, lets shutdown them all", m) |
| 163 | for i := len(m.runners) - 2; i >= 0; i-- { |
| 164 | //u.Debugf("%p sending close??: %v %T", m, i, m.runners[i]) |
| 165 | m.runners[i].Close() |
| 166 | //u.Debugf("%p after close??: %v %T", m, i, m.runners[i]) |
| 167 | } |
| 168 | } |
| 169 | }(i) |
| 170 | } |
| 171 | |
| 172 | wg.Wait() // block until all tasks have finished |
| 173 | //u.Debugf("%p exit TaskSequential Run(): %q", m, m.Name) |
| 174 | return |
| 175 | } |