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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
package factory
import (
"errors"
"fmt"
"os"
"os/exec"
"strings"
"time"
"github.com/spiral/roadrunner/v2"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/util"
)
// AppConfig config combines factory, pool and cmd configurations.
type AppConfig struct {
Command string
User string
Group string
Env Env
// Listen defines connection method and factory to be used to connect to workers:
// "pipes", "tcp://:6001", "unix://rr.sock"
// This config section must not change on re-configuration.
Relay string
// RelayTimeout defines for how long socket factory will be waiting for worker connection. This config section
// must not change on re-configuration.
RelayTimeout time.Duration
}
type App struct {
cfg AppConfig
configProvider config.Provider
factory roadrunner.Factory
}
func (app *App) Init(provider config.Provider) error {
app.cfg = AppConfig{}
app.configProvider = provider
return nil
}
func (app *App) Configure() error {
err := app.configProvider.UnmarshalKey("app", &app.cfg)
if err != nil {
return err
}
if app.cfg.Relay == "" {
app.cfg.Relay = "pipes"
}
return nil
}
func (app *App) Close() error {
return nil
}
func (app *App) NewCmd(env Env) (func() *exec.Cmd, error) {
var cmdArgs []string
// create command according to the config
cmdArgs = append(cmdArgs, strings.Split(app.cfg.Command, " ")...)
return func() *exec.Cmd {
cmd := exec.Command(cmdArgs[0], cmdArgs[1:]...)
util.IsolateProcess(cmd)
// if user is not empty, and OS is linux or macos
// execute php worker from that particular user
if app.cfg.User != "" {
err := util.ExecuteFromUser(cmd, app.cfg.User)
if err != nil {
return nil
}
}
cmd.Env = app.setEnv(env)
return cmd
}, nil
}
// todo ENV unused
func (app *App) NewFactory(env Env) (roadrunner.Factory, error) {
if app.cfg.Relay == "" || app.cfg.Relay == "pipes" {
return roadrunner.NewPipeFactory(), nil
}
dsn := strings.Split(app.cfg.Relay, "://")
if len(dsn) != 2 {
return nil, errors.New("invalid DSN (tcp://:6001, unix://file.sock)")
}
lsn, err := util.CreateListener(app.cfg.Relay)
if err != nil {
return nil, err
}
switch dsn[0] {
// sockets group
case "unix":
return roadrunner.NewSocketServer(lsn, app.cfg.RelayTimeout), nil
case "tcp":
return roadrunner.NewSocketServer(lsn, app.cfg.RelayTimeout), nil
default:
return nil, errors.New("invalid DSN (tcp://:6001, unix://file.sock)")
}
}
func (app *App) Serve() chan error {
errCh := make(chan error)
return errCh
}
func (app *App) Stop() error {
return nil
}
func (app *App) setEnv(e Env) []string {
env := append(os.Environ(), fmt.Sprintf("RR_RELAY=%s", app.cfg.Relay))
for k, v := range e {
env = append(env, fmt.Sprintf("%s=%s", strings.ToUpper(k), v))
}
return env
}
|
