1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
package roadrunner
import (
"fmt"
"io"
"os/exec"
"github.com/pkg/errors"
"github.com/spiral/goridge/v2"
)
// PipeFactory connects to workers using standard
// streams (STDIN, STDOUT pipes).
type PipeFactory struct {
}
// NewPipeFactory returns new factory instance and starts
// listening
func NewPipeFactory() *PipeFactory {
return &PipeFactory{}
}
// SpawnWorker creates new worker and connects it to goridge relay,
// method Wait() must be handled on level above.
func (f *PipeFactory) SpawnWorker(cmd *exec.Cmd) (w *Worker, err error) {
if w, err = newWorker(cmd); err != nil {
return nil, err
}
var (
in io.ReadCloser
out io.WriteCloser
)
if in, err = cmd.StdoutPipe(); err != nil {
return nil, err
}
if out, err = cmd.StdinPipe(); err != nil {
return nil, err
}
w.rl = goridge.NewPipeRelay(in, out)
if err := w.start(); err != nil {
return nil, errors.Wrap(err, "process error")
}
if pid, err := fetchPID(w.rl); pid != *w.Pid {
go func(w *Worker) {
err := w.Kill()
if err != nil {
// there is no logger here, how to handle error in goroutines ?
fmt.Printf("error killing the worker with PID number %d, Created: %s", w.Pid, w.Created)
}
}(w)
if wErr := w.Wait(); wErr != nil {
if _, ok := wErr.(*exec.ExitError); ok {
// error might be nil here
if err != nil {
err = errors.Wrap(wErr, err.Error())
}
} else {
err = wErr
}
}
return nil, errors.Wrap(err, "unable to connect to worker")
}
w.state.set(StateReady)
return w, nil
}
// Close the factory.
func (f *PipeFactory) Close() error {
return nil
}
|
