diff options
Diffstat (limited to 'plugins')
39 files changed, 984 insertions, 97 deletions
diff --git a/plugins/broadcast/interface.go b/plugins/broadcast/interface.go index 46709d71..eda3572f 100644 --- a/plugins/broadcast/interface.go +++ b/plugins/broadcast/interface.go @@ -1,6 +1,6 @@ package broadcast -import "github.com/spiral/roadrunner/v2/pkg/pubsub" +import "github.com/spiral/roadrunner/v2/common/pubsub" type Broadcaster interface { GetDriver(key string) (pubsub.SubReader, error) diff --git a/plugins/broadcast/plugin.go b/plugins/broadcast/plugin.go index 6ddef806..889dc2fa 100644 --- a/plugins/broadcast/plugin.go +++ b/plugins/broadcast/plugin.go @@ -7,7 +7,7 @@ import ( "github.com/google/uuid" endure "github.com/spiral/endure/pkg/container" "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/config" "github.com/spiral/roadrunner/v2/plugins/logger" ) diff --git a/plugins/broadcast/rpc.go b/plugins/broadcast/rpc.go index 2ee211f8..475076a0 100644 --- a/plugins/broadcast/rpc.go +++ b/plugins/broadcast/rpc.go @@ -2,7 +2,7 @@ package broadcast import ( "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/logger" websocketsv1 "github.com/spiral/roadrunner/v2/proto/websockets/v1beta" ) diff --git a/plugins/http/plugin.go b/plugins/http/plugin.go index bec01ac3..fb174792 100644 --- a/plugins/http/plugin.go +++ b/plugins/http/plugin.go @@ -143,7 +143,7 @@ func (p *Plugin) Serve() chan error { func (p *Plugin) serve(errCh chan error) { var err error const op = errors.Op("http_plugin_serve") - p.pool, err = p.server.NewWorkerPool(context.Background(), pool.Config{ + p.pool, err = p.server.NewWorkerPool(context.Background(), &pool.Config{ Debug: p.cfg.Pool.Debug, NumWorkers: p.cfg.Pool.NumWorkers, MaxJobs: p.cfg.Pool.MaxJobs, @@ -323,7 +323,7 @@ func (p *Plugin) Reset() error { p.pool = nil var err error - p.pool, err = p.server.NewWorkerPool(context.Background(), pool.Config{ + p.pool, err = p.server.NewWorkerPool(context.Background(), &pool.Config{ Debug: p.cfg.Pool.Debug, NumWorkers: p.cfg.Pool.NumWorkers, MaxJobs: p.cfg.Pool.MaxJobs, diff --git a/plugins/jobs/brokers/ephemeral/broker.go b/plugins/jobs/brokers/ephemeral/broker.go new file mode 100644 index 00000000..905f5409 --- /dev/null +++ b/plugins/jobs/brokers/ephemeral/broker.go @@ -0,0 +1,30 @@ +package ephemeral + +import ( + priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue" + "github.com/spiral/roadrunner/v2/plugins/jobs/pipeline" + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" +) + +type JobBroker struct { +} + +func NewJobBroker(q priorityqueue.Queue) (*JobBroker, error) { + return &JobBroker{}, nil +} + +func (j *JobBroker) Push(pipeline *pipeline.Pipeline, job *structs.Job) (string, error) { + panic("implement me") +} + +func (j *JobBroker) Stat() { + panic("implement me") +} + +func (j *JobBroker) Consume(pipeline *pipeline.Pipeline) { + panic("implement me") +} + +func (j *JobBroker) Register(pipeline *pipeline.Pipeline) { + panic("implement me") +} diff --git a/plugins/jobs/brokers/ephemeral/config.go b/plugins/jobs/brokers/ephemeral/config.go new file mode 100644 index 00000000..847b63ea --- /dev/null +++ b/plugins/jobs/brokers/ephemeral/config.go @@ -0,0 +1 @@ +package ephemeral diff --git a/plugins/jobs/brokers/ephemeral/plugin.go b/plugins/jobs/brokers/ephemeral/plugin.go new file mode 100644 index 00000000..84cc871b --- /dev/null +++ b/plugins/jobs/brokers/ephemeral/plugin.go @@ -0,0 +1,28 @@ +package ephemeral + +import ( + priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue" + "github.com/spiral/roadrunner/v2/plugins/jobs" + "github.com/spiral/roadrunner/v2/plugins/logger" +) + +const ( + PluginName string = "ephemeral" +) + +type Plugin struct { + log logger.Logger +} + +func (p *Plugin) Init(log logger.Logger) error { + p.log = log + return nil +} + +func (p *Plugin) Name() string { + return PluginName +} + +func (p *Plugin) InitJobBroker(q priorityqueue.Queue) (jobs.Consumer, error) { + return NewJobBroker(q) +} diff --git a/plugins/jobs/config.go b/plugins/jobs/config.go new file mode 100644 index 00000000..bb042ec9 --- /dev/null +++ b/plugins/jobs/config.go @@ -0,0 +1,71 @@ +package jobs + +import ( + "github.com/spiral/errors" + poolImpl "github.com/spiral/roadrunner/v2/pkg/pool" + "github.com/spiral/roadrunner/v2/plugins/jobs/dispatcher" + "github.com/spiral/roadrunner/v2/plugins/jobs/pipeline" + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" +) + +// Config defines settings for job broker, workers and job-pipeline mapping. +type Config struct { + // Workers configures roadrunner server and worker busy. + // Workers *roadrunner.ServerConfig + poolCfg *poolImpl.Config + + // Dispatch defines where and how to match jobs. + Dispatch map[string]*structs.Options + + // Pipelines defines mapping between PHP job pipeline and associated job broker. + Pipelines map[string]*pipeline.Pipeline + + // Consuming specifies names of pipelines to be consumed on service start. + Consume []string + + // parent config for broken options. + pipelines pipeline.Pipelines + route dispatcher.Dispatcher +} + +func (c *Config) InitDefaults() error { + const op = errors.Op("config_init_defaults") + var err error + c.pipelines, err = pipeline.InitPipelines(c.Pipelines) + if err != nil { + return errors.E(op, err) + } + + if c.poolCfg == nil { + c.poolCfg = &poolImpl.Config{} + } + + c.poolCfg.InitDefaults() + + return nil +} + +// MatchPipeline locates the pipeline associated with the job. +func (c *Config) MatchPipeline(job *structs.Job) (*pipeline.Pipeline, *structs.Options, error) { + const op = errors.Op("config_match_pipeline") + opt := c.route.Match(job) + + pipe := "" + if job.Options != nil { + pipe = job.Options.Pipeline + } + + if pipe == "" && opt != nil { + pipe = opt.Pipeline + } + + if pipe == "" { + return nil, nil, errors.E(op, errors.Errorf("unable to locate pipeline for `%s`", job.Job)) + } + + if p := c.pipelines.Get(pipe); p != nil { + return p, opt, nil + } + + return nil, nil, errors.E(op, errors.Errorf("undefined pipeline `%s`", pipe)) +} diff --git a/plugins/jobs/dispatcher/dispatcher.go b/plugins/jobs/dispatcher/dispatcher.go new file mode 100644 index 00000000..e73e7b74 --- /dev/null +++ b/plugins/jobs/dispatcher/dispatcher.go @@ -0,0 +1,49 @@ +package dispatcher + +import ( + "strings" + + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" +) + +var separators = []string{"/", "-", "\\"} + +// Dispatcher provides ability to automatically locate the pipeline for the specific job +// and update job options (if none set). +type Dispatcher map[string]*structs.Options + +// pre-compile patterns +func initDispatcher(routes map[string]*structs.Options) Dispatcher { + dispatcher := make(Dispatcher) + for pattern, opts := range routes { + pattern = strings.ToLower(pattern) + pattern = strings.Trim(pattern, "-.*") + + for _, s := range separators { + pattern = strings.ReplaceAll(pattern, s, ".") + } + + dispatcher[pattern] = opts + } + + return dispatcher +} + +// Match clarifies target job pipeline and other job options. Can return nil. +func (dispatcher Dispatcher) Match(job *structs.Job) (found *structs.Options) { + var best = 0 + + jobName := strings.ToLower(job.Job) + for pattern, opts := range dispatcher { + if strings.HasPrefix(jobName, pattern) && len(pattern) > best { + found = opts + best = len(pattern) + } + } + + if best == 0 { + return nil + } + + return found +} diff --git a/plugins/jobs/dispatcher/dispatcher_test.go b/plugins/jobs/dispatcher/dispatcher_test.go new file mode 100644 index 00000000..e584bda8 --- /dev/null +++ b/plugins/jobs/dispatcher/dispatcher_test.go @@ -0,0 +1,55 @@ +package dispatcher + +import ( + "testing" + + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" + "github.com/stretchr/testify/assert" +) + +func Test_Map_All(t *testing.T) { + m := initDispatcher(map[string]*structs.Options{"default": {Pipeline: "default"}}) + assert.Equal(t, "default", m.Match(&structs.Job{Job: "default"}).Pipeline) +} + +func Test_Map_Miss(t *testing.T) { + m := initDispatcher(map[string]*structs.Options{"some.*": {Pipeline: "default"}}) + + assert.Nil(t, m.Match(&structs.Job{Job: "miss"})) +} + +func Test_Map_Best(t *testing.T) { + m := initDispatcher(map[string]*structs.Options{ + "some.*": {Pipeline: "default"}, + "some.other.*": {Pipeline: "other"}, + }) + + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some"}).Pipeline) + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some.any"}).Pipeline) + assert.Equal(t, "other", m.Match(&structs.Job{Job: "some.other"}).Pipeline) + assert.Equal(t, "other", m.Match(&structs.Job{Job: "some.other.job"}).Pipeline) +} + +func Test_Map_BestUpper(t *testing.T) { + m := initDispatcher(map[string]*structs.Options{ + "some.*": {Pipeline: "default"}, + "some.Other.*": {Pipeline: "other"}, + }) + + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some"}).Pipeline) + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some.any"}).Pipeline) + assert.Equal(t, "other", m.Match(&structs.Job{Job: "some.OTHER"}).Pipeline) + assert.Equal(t, "other", m.Match(&structs.Job{Job: "Some.other.job"}).Pipeline) +} + +func Test_Map_BestReversed(t *testing.T) { + m := initDispatcher(map[string]*structs.Options{ + "some.*": {Pipeline: "default"}, + "some.other.*": {Pipeline: "other"}, + }) + + assert.Equal(t, "other", m.Match(&structs.Job{Job: "some.other.job"}).Pipeline) + assert.Equal(t, "other", m.Match(&structs.Job{Job: "some.other"}).Pipeline) + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some.any"}).Pipeline) + assert.Equal(t, "default", m.Match(&structs.Job{Job: "some"}).Pipeline) +} diff --git a/plugins/jobs/doc/jobs_arch.drawio b/plugins/jobs/doc/jobs_arch.drawio new file mode 100644 index 00000000..d452d345 --- /dev/null +++ b/plugins/jobs/doc/jobs_arch.drawio @@ -0,0 +1 @@ +<mxfile host="Electron" modified="2021-06-23T11:05:51.495Z" agent="5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) draw.io/14.6.13 Chrome/91.0.4472.106 Electron/13.1.4 Safari/537.36" etag="_SQi43bjgSX-nT5V1Ksg" version="14.6.13" type="device"><diagram id="AFQlLRRq6yGg9IpTzkrs" name="Page-1">7VvrU9s4EP9rMkOZIeO3nY+E8Givvebgblrum2IriVvbcmUlJPz1t7Llp5yQgpNAr1BKtHpYu/vbh1amp1+Eq2uK4vkn4uGgpyneqqePepqmDiwFfnHKWlBUS88oM+p7glYS7vxHLIhi4mzhezipDWSEBMyP60SXRBF2WY2GKCUP9WFTEtSfGqMZlgh3Lgpk6hffY/N8d/BV9txgfzZnUleI8vGCkMyRRx4qJP2yp19QQlj2KVxd4IALMBdNNu9qQ2+xN4ojtsuEr1G0ulFJiJePLLKv/8L/Ku6ZZmTLLFGwEEyL3bJ1LgVKFpGH+SpKTx8+zH2G72Lk8t4HUDzQ5iwMoKXCx6kfBBckIDSdq3sIO1MX6Amj5Duu9FiugydT6JH5EKwtMWV4VSEJvq4xMMHoGoaIXt0RMhZAG9g5gh5KtWn5oHlVYbl2kIDKrFi8FCV8ENL8GclKcsQegEs0CWVzMiMRCi5L6rAu6XLMR0JiId9vmLG1sBS0YKQufRAhXX/l8/tm3ryv9o1WYvGstRatjTpIyIK6eBubgk+G6AyzLQPFglwIW1VKcYCYv6xbYJt2xNQx8WHPBRQMw6xDwbQbGs52KqZV7UVaqQEqc9BYKROOtNI5pWhdGRbzAYkEqILn52NMPwrGVj6rQAxa9zmm4HMJMN5YV9F2UFyqO8LSOBQsG2DSrd1g2RVUcp9a+vnbMayknMHP+J+7G1g0+wUbWkdut0HAxI5ntAUBR5voltVRELAbIjY0OQioRksQ0PcVAwZHtc+qdRa2+nP2qezNPo1dw4ZyGAPVmugpcopDGehvZ/5ysFgH8ub6oO8MLNNUbc1QdXtQh47WTCr3DB05h/+EmDvnnv39+PLj+z8vO3XnWAWHbre584Fl66grd97M6fU2d661uPO9pfT2MS1Urblz+7X5c3NHE+3cQl+kUefN+Vy1ps5Su0dTaOfnuhcp1JRcYU+zAtj/MK7p2fqx4HWXYQjs+VFPP4deJV7B/6nQlIx+xrhKeZ9R6QMHxs5Q4M/EPBdki2m928MuocAvEWM4ZGjgR7h8NHyaid/pBic54QOZgFjO7xhduGxBcT4ABDJpTgJa3KTNaZPyLN4DPGVZp8M7m9uGXaYzRuna1I9mBSPF88doHRDkPTnuc8wllRTjinYbkw2D5RHFd1FwnilklKpsKNQzypgYEhg1DdLqGw9mPKaRiAkLVTXRvkKhH3DLucHBEvNVt8bBqcm/W2tb6ZdYtULPvrqJj5baLE8oLfHRaomPzTJGZ8Zn/QLGl0Pv1dvf2I9xxtMTxjUCx7t+ctQ5YziMWWmDaa66mcl9sHTLUVjfbmUXE0msh9rX336IyYI9Z1O/nVXa60i11F2dlaruy1upqqSeo6XzO2fz2qHKp8qO6V/n9dOX6VSRQtAtWC7upwfik29k8q7To/DUcbHber01cUzD3KqD3a1HNZtH4cGOR+G9VTZVTRI0xRAso2r21hA0j811aeaOTwRxfdj0j6HveZnd4cR/RJN0KS5VcacC65rDnjnia4GpJcJHSgqJSIS70YUmlSUcWReDQ1YlVLn8E0Nu0NN4eZ/8yrowmnahWbIu2m5996cL+fxZydPOIC9AIfcoIjuA/CMGoYmczBz5XPRT7nvsYc8e/V+01mJB6mFN6KiVvddde89r6k8XgpyuU4FnFd/VZszb972NfOoNMTyRi7s0/U1++A3W4dXjJx/HvVd93ebq7GiuucaObK7Kge/KcvlUzPUa8+0OuQ3xQtKUkpALcI5T5md+ApEV8yLmJB3CSyRLHwn77mfzTjo+V0ynWvu5wrMmlrkv01baInGbaTdfXunurTn5ADdeJPOTMqPNznAKWCc8vqK3DqV/nJcWncGu0jf2Jn35LaEN4v01MlGriX9D1oB5yERUk4/VkujJgvGk4qJ451mRa3fwfcUfO5xR5Pm47BPCawC+GN4qaQ8l88KScq1+RBMcjEFLWZl9NCGMgd+U1V4vcRboqFpfS3Uzj+X8kSiJM0an/orvIwMMppdLnOEmxcgcxXxCuJrxl9H76CEx+oskfVYXUFHUOlRsXYaK7chQyWndQ0WXPeXNGAhfPt/+cXkrwSYXUEyJi5Pkaac4Qe73WepGP2eI2xCqDlICM/WGqdqG0zd3dJcdVBsfran3oDtXxvfw/P56sRrbg5szWQOHu+9qucjKSguB3CrLCo36Q6O14UajuPQCl5MswjRJOslyoXdHuhtrFZV0mbNNjukoSYp3DLGT07xkw/OMk1NIOJI+78jyD0wpqfHdzRXUvjhKs6fTsgjFWTj9kOVQJ/ll4BvkS2Cxpqw3svVbcaaoA62nX4l/oBjwogHEPH7USEiwxMkG7jb4eJrGy6ozl+8U27x/B45aLVzwtsuK4l3tmp/uIq9tddS/6wUbRdPymlfruM6Ley9SaEuZPf3jhZccArsAv9UEv9F2plNawN9FsWyb4iqi+tHnIY3nEnMEZ4jgFYit+S7TPi84oVn+0WVWVir/fFW//A8=</diagram></mxfile>
\ No newline at end of file diff --git a/plugins/jobs/interface.go b/plugins/jobs/interface.go new file mode 100644 index 00000000..bb0e8c50 --- /dev/null +++ b/plugins/jobs/interface.go @@ -0,0 +1,19 @@ +package jobs + +import ( + priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue" + "github.com/spiral/roadrunner/v2/plugins/jobs/pipeline" + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" +) + +// Consumer todo naming +type Consumer interface { + Push(*pipeline.Pipeline, *structs.Job) (string, error) + Stat() + Consume(*pipeline.Pipeline) + Register(*pipeline.Pipeline) +} + +type Broker interface { + InitJobBroker(queue priorityqueue.Queue) (Consumer, error) +} diff --git a/plugins/jobs/pipeline/pipeline.go b/plugins/jobs/pipeline/pipeline.go new file mode 100644 index 00000000..f27f6ede --- /dev/null +++ b/plugins/jobs/pipeline/pipeline.go @@ -0,0 +1,172 @@ +package pipeline + +import ( + "time" + + "github.com/spiral/errors" +) + +// Pipelines is list of Pipeline. + +type Pipelines []*Pipeline + +func InitPipelines(pipes map[string]*Pipeline) (Pipelines, error) { + const op = errors.Op("pipeline_init") + out := make(Pipelines, 0) + + for name, pipe := range pipes { + if pipe.Broker() == "" { + return nil, errors.E(op, errors.Errorf("found the pipeline without defined broker")) + } + + p := pipe.With("name", name) + out = append(out, &p) + } + + return out, nil +} + +// Reverse returns pipelines in reversed order. +func (ps Pipelines) Reverse() Pipelines { + out := make(Pipelines, len(ps)) + + for i, p := range ps { + out[len(ps)-i-1] = p + } + + return out +} + +// Broker return pipelines associated with specific broker. +func (ps Pipelines) Broker(broker string) Pipelines { + out := make(Pipelines, 0) + + for _, p := range ps { + if p.Broker() != broker { + continue + } + + out = append(out, p) + } + + return out +} + +// Names returns only pipelines with specified names. +func (ps Pipelines) Names(only ...string) Pipelines { + out := make(Pipelines, 0) + + for _, name := range only { + for _, p := range ps { + if p.Name() == name { + out = append(out, p) + } + } + } + + return out +} + +// Get returns pipeline by it'svc name. +func (ps Pipelines) Get(name string) *Pipeline { + // possibly optimize + for _, p := range ps { + if p.Name() == name { + return p + } + } + + return nil +} + +// Pipeline defines pipeline options. +type Pipeline map[string]interface{} + +// With pipeline value. Immutable. +func (p Pipeline) With(name string, value interface{}) Pipeline { + out := make(map[string]interface{}) + for k, v := range p { + out[k] = v + } + out[name] = value + + return out +} + +// Name returns pipeline name. +func (p Pipeline) Name() string { + return p.String("name", "") +} + +// Broker associated with the pipeline. +func (p Pipeline) Broker() string { + return p.String("broker", "") +} + +// Has checks if value presented in pipeline. +func (p Pipeline) Has(name string) bool { + if _, ok := p[name]; ok { + return true + } + + return false +} + +// Map must return nested map value or empty config. +func (p Pipeline) Map(name string) Pipeline { + out := make(map[string]interface{}) + + if value, ok := p[name]; ok { + if m, ok := value.(map[string]interface{}); ok { + for k, v := range m { + out[k] = v + } + } + } + + return out +} + +// Bool must return option value as string or return default value. +func (p Pipeline) Bool(name string, d bool) bool { + if value, ok := p[name]; ok { + if b, ok := value.(bool); ok { + return b + } + } + + return d +} + +// String must return option value as string or return default value. +func (p Pipeline) String(name string, d string) string { + if value, ok := p[name]; ok { + if str, ok := value.(string); ok { + return str + } + } + + return d +} + +// Integer must return option value as string or return default value. +func (p Pipeline) Integer(name string, d int) int { + if value, ok := p[name]; ok { + if str, ok := value.(int); ok { + return str + } + } + + return d +} + +// Duration must return option value as time.Duration (seconds) or return default value. +func (p Pipeline) Duration(name string, d time.Duration) time.Duration { + if value, ok := p[name]; ok { + if str, ok := value.(int); ok { + return time.Second * time.Duration(str) + } + } + + return d +} diff --git a/plugins/jobs/pipeline/pipeline_test.go b/plugins/jobs/pipeline/pipeline_test.go new file mode 100644 index 00000000..f03dcbb8 --- /dev/null +++ b/plugins/jobs/pipeline/pipeline_test.go @@ -0,0 +1,90 @@ +package pipeline + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestPipeline_Map(t *testing.T) { + pipe := Pipeline{"options": map[string]interface{}{"ttl": 10}} + + assert.Equal(t, 10, pipe.Map("options").Integer("ttl", 0)) + assert.Equal(t, 0, pipe.Map("other").Integer("ttl", 0)) +} + +func TestPipeline_MapString(t *testing.T) { + pipe := Pipeline{"options": map[string]interface{}{"alias": "default"}} + + assert.Equal(t, "default", pipe.Map("options").String("alias", "")) + assert.Equal(t, "", pipe.Map("other").String("alias", "")) +} + +func TestPipeline_Bool(t *testing.T) { + pipe := Pipeline{"value": true} + + assert.Equal(t, true, pipe.Bool("value", false)) + assert.Equal(t, true, pipe.Bool("other", true)) +} + +func TestPipeline_String(t *testing.T) { + pipe := Pipeline{"value": "value"} + + assert.Equal(t, "value", pipe.String("value", "")) + assert.Equal(t, "value", pipe.String("other", "value")) +} + +func TestPipeline_Integer(t *testing.T) { + pipe := Pipeline{"value": 1} + + assert.Equal(t, 1, pipe.Integer("value", 0)) + assert.Equal(t, 1, pipe.Integer("other", 1)) +} + +func TestPipeline_Duration(t *testing.T) { + pipe := Pipeline{"value": 1} + + assert.Equal(t, time.Second, pipe.Duration("value", 0)) + assert.Equal(t, time.Second, pipe.Duration("other", time.Second)) +} + +func TestPipeline_Has(t *testing.T) { + pipe := Pipeline{"options": map[string]interface{}{"ttl": 10}} + + assert.Equal(t, true, pipe.Has("options")) + assert.Equal(t, false, pipe.Has("other")) +} + +func TestPipeline_FilterBroker(t *testing.T) { + pipes := Pipelines{ + &Pipeline{"name": "first", "broker": "a"}, + &Pipeline{"name": "second", "broker": "a"}, + &Pipeline{"name": "third", "broker": "b"}, + &Pipeline{"name": "forth", "broker": "b"}, + } + + filtered := pipes.Names("first", "third") + assert.True(t, len(filtered) == 2) + + assert.Equal(t, "a", filtered[0].Broker()) + assert.Equal(t, "b", filtered[1].Broker()) + + filtered = pipes.Names("first", "third").Reverse() + assert.True(t, len(filtered) == 2) + + assert.Equal(t, "a", filtered[1].Broker()) + assert.Equal(t, "b", filtered[0].Broker()) + + filtered = pipes.Broker("a") + assert.True(t, len(filtered) == 2) + + assert.Equal(t, "first", filtered[0].Name()) + assert.Equal(t, "second", filtered[1].Name()) + + filtered = pipes.Broker("a").Reverse() + assert.True(t, len(filtered) == 2) + + assert.Equal(t, "first", filtered[1].Name()) + assert.Equal(t, "second", filtered[0].Name()) +} diff --git a/plugins/jobs/plugin.go b/plugins/jobs/plugin.go new file mode 100644 index 00000000..90932edd --- /dev/null +++ b/plugins/jobs/plugin.go @@ -0,0 +1,148 @@ +package jobs + +import ( + "context" + "fmt" + + endure "github.com/spiral/endure/pkg/container" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/pkg/events" + "github.com/spiral/roadrunner/v2/pkg/pool" + priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" + "github.com/spiral/roadrunner/v2/plugins/logger" + "github.com/spiral/roadrunner/v2/plugins/server" +) + +const ( + // RrJobs env variable + RrJobs string = "rr_jobs" + PluginName string = "jobs" +) + +type Plugin struct { + cfg *Config + log logger.Logger + + workersPool pool.Pool + server server.Server + + brokers map[string]Broker + consumers map[string]Consumer + + events events.Handler + + // priority queue implementation + queue priorityqueue.Queue +} + +func testListener(data interface{}) { + fmt.Println(data) +} + +func (p *Plugin) Init(cfg config.Configurer, log logger.Logger, server server.Server) error { + const op = errors.Op("jobs_plugin_init") + if !cfg.Has(PluginName) { + return errors.E(op, errors.Disabled) + } + + err := cfg.UnmarshalKey(PluginName, &p.cfg) + if err != nil { + return errors.E(op, err) + } + + err = p.cfg.InitDefaults() + if err != nil { + return errors.E(op, err) + } + + p.server = server + p.events = events.NewEventsHandler() + p.events.AddListener(testListener) + p.brokers = make(map[string]Broker) + p.consumers = make(map[string]Consumer) + + // initialize priority queue + p.queue = priorityqueue.NewPriorityQueue() + p.log = log + + return nil +} + +func (p *Plugin) Serve() chan error { + errCh := make(chan error, 1) + + for name := range p.brokers { + jb, err := p.brokers[name].InitJobBroker(p.queue) + if err != nil { + errCh <- err + return errCh + } + + p.consumers[name] = jb + } + + var err error + p.workersPool, err = p.server.NewWorkerPool(context.Background(), p.cfg.poolCfg, map[string]string{RrJobs: "true"}, testListener) + if err != nil { + errCh <- err + return errCh + } + + // initialize sub-plugins + // provide a queue to them + // start consume loop + // start resp loop + + return errCh +} + +func (p *Plugin) Stop() error { + return nil +} + +func (p *Plugin) Collects() []interface{} { + return []interface{}{ + p.CollectMQBrokers, + } +} + +func (p *Plugin) CollectMQBrokers(name endure.Named, c Broker) { + p.brokers[name.Name()] = c +} + +func (p *Plugin) Available() {} + +func (p *Plugin) Name() string { + return PluginName +} + +func (p *Plugin) Push(j *structs.Job) (string, error) { + pipe, pOpts, err := p.cfg.MatchPipeline(j) + if err != nil { + panic(err) + } + + if pOpts != nil { + j.Options.Merge(pOpts) + } + + broker, ok := p.consumers[pipe.Broker()] + if !ok { + panic("broker not found") + } + + id, err := broker.Push(pipe, j) + if err != nil { + panic(err) + } + + // p.events.Push() + + return id, nil +} + +func (p *Plugin) RPC() interface{} { + return &rpc{log: p.log} +} diff --git a/plugins/jobs/rpc.go b/plugins/jobs/rpc.go new file mode 100644 index 00000000..e77cda59 --- /dev/null +++ b/plugins/jobs/rpc.go @@ -0,0 +1,20 @@ +package jobs + +import ( + "github.com/spiral/roadrunner/v2/plugins/jobs/structs" + "github.com/spiral/roadrunner/v2/plugins/logger" +) + +type rpc struct { + log logger.Logger + p *Plugin +} + +func (r *rpc) Push(j *structs.Job, idRet *string) error { + id, err := r.p.Push(j) + if err != nil { + panic(err) + } + *idRet = id + return nil +} diff --git a/plugins/jobs/structs/job.go b/plugins/jobs/structs/job.go new file mode 100644 index 00000000..2e394543 --- /dev/null +++ b/plugins/jobs/structs/job.go @@ -0,0 +1,35 @@ +package structs + +import ( + json "github.com/json-iterator/go" + "github.com/spiral/roadrunner/v2/utils" +) + +// Job carries information about single job. +type Job struct { + // Job contains name of job broker (usually PHP class). + Job string `json:"job"` + + // Payload is string data (usually JSON) passed to Job broker. + Payload string `json:"payload"` + + // Options contains set of PipelineOptions specific to job execution. Can be empty. + Options *Options `json:"options,omitempty"` +} + +// Body packs job payload into binary payload. +func (j *Job) Body() []byte { + return utils.AsBytes(j.Payload) +} + +// Context packs job context (job, id) into binary payload. +func (j *Job) Context(id string) []byte { + ctx, _ := json.Marshal( + struct { + ID string `json:"id"` + Job string `json:"job"` + }{ID: id, Job: j.Job}, + ) + + return ctx +} diff --git a/plugins/jobs/structs/job_options.go b/plugins/jobs/structs/job_options.go new file mode 100644 index 00000000..1507d053 --- /dev/null +++ b/plugins/jobs/structs/job_options.go @@ -0,0 +1,70 @@ +package structs + +import "time" + +// Options carry information about how to handle given job. +type Options struct { + // Pipeline manually specified pipeline. + Pipeline string `json:"pipeline,omitempty"` + + // Delay defines time duration to delay execution for. Defaults to none. + Delay int `json:"delay,omitempty"` + + // Attempts define maximum job retries. Attention, value 1 will only allow job to execute once (without retry). + // Minimum valuable value is 2. + Attempts int `json:"maxAttempts,omitempty"` + + // RetryDelay defines for how long job should be waiting until next retry. Defaults to none. + RetryDelay int `json:"retryDelay,omitempty"` + + // Reserve defines for how broker should wait until treating job are failed. Defaults to 30 min. + Timeout int `json:"timeout,omitempty"` +} + +// Merge merges job options. +func (o *Options) Merge(from *Options) { + if o.Pipeline == "" { + o.Pipeline = from.Pipeline + } + + if o.Attempts == 0 { + o.Attempts = from.Attempts + } + + if o.Timeout == 0 { + o.Timeout = from.Timeout + } + + if o.RetryDelay == 0 { + o.RetryDelay = from.RetryDelay + } + + if o.Delay == 0 { + o.Delay = from.Delay + } +} + +// CanRetry must return true if broker is allowed to re-run the job. +func (o *Options) CanRetry(attempt int) bool { + // Attempts 1 and 0 has identical effect + return o.Attempts > (attempt + 1) +} + +// RetryDuration returns retry delay duration in a form of time.Duration. +func (o *Options) RetryDuration() time.Duration { + return time.Second * time.Duration(o.RetryDelay) +} + +// DelayDuration returns delay duration in a form of time.Duration. +func (o *Options) DelayDuration() time.Duration { + return time.Second * time.Duration(o.Delay) +} + +// TimeoutDuration returns timeout duration in a form of time.Duration. +func (o *Options) TimeoutDuration() time.Duration { + if o.Timeout == 0 { + return 30 * time.Minute + } + + return time.Second * time.Duration(o.Timeout) +} diff --git a/plugins/jobs/structs/job_options_test.go b/plugins/jobs/structs/job_options_test.go new file mode 100644 index 00000000..18702394 --- /dev/null +++ b/plugins/jobs/structs/job_options_test.go @@ -0,0 +1,110 @@ +package structs + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestOptions_CanRetry(t *testing.T) { + opts := &Options{Attempts: 0} + + assert.False(t, opts.CanRetry(0)) + assert.False(t, opts.CanRetry(1)) +} + +func TestOptions_CanRetry_SameValue(t *testing.T) { + opts := &Options{Attempts: 1} + + assert.False(t, opts.CanRetry(0)) + assert.False(t, opts.CanRetry(1)) +} + +func TestOptions_CanRetry_Value(t *testing.T) { + opts := &Options{Attempts: 2} + + assert.True(t, opts.CanRetry(0)) + assert.False(t, opts.CanRetry(1)) + assert.False(t, opts.CanRetry(2)) +} + +func TestOptions_CanRetry_Value3(t *testing.T) { + opts := &Options{Attempts: 3} + + assert.True(t, opts.CanRetry(0)) + assert.True(t, opts.CanRetry(1)) + assert.False(t, opts.CanRetry(2)) +} + +func TestOptions_RetryDuration(t *testing.T) { + opts := &Options{RetryDelay: 0} + assert.Equal(t, time.Duration(0), opts.RetryDuration()) +} + +func TestOptions_RetryDuration2(t *testing.T) { + opts := &Options{RetryDelay: 1} + assert.Equal(t, time.Second, opts.RetryDuration()) +} + +func TestOptions_DelayDuration(t *testing.T) { + opts := &Options{Delay: 0} + assert.Equal(t, time.Duration(0), opts.DelayDuration()) +} + +func TestOptions_DelayDuration2(t *testing.T) { + opts := &Options{Delay: 1} + assert.Equal(t, time.Second, opts.DelayDuration()) +} + +func TestOptions_TimeoutDuration(t *testing.T) { + opts := &Options{Timeout: 0} + assert.Equal(t, time.Minute*30, opts.TimeoutDuration()) +} + +func TestOptions_TimeoutDuration2(t *testing.T) { + opts := &Options{Timeout: 1} + assert.Equal(t, time.Second, opts.TimeoutDuration()) +} + +func TestOptions_Merge(t *testing.T) { + opts := &Options{} + + opts.Merge(&Options{ + Pipeline: "pipeline", + Delay: 2, + Timeout: 1, + Attempts: 1, + RetryDelay: 1, + }) + + assert.Equal(t, "pipeline", opts.Pipeline) + assert.Equal(t, 1, opts.Attempts) + assert.Equal(t, 2, opts.Delay) + assert.Equal(t, 1, opts.Timeout) + assert.Equal(t, 1, opts.RetryDelay) +} + +func TestOptions_MergeKeepOriginal(t *testing.T) { + opts := &Options{ + Pipeline: "default", + Delay: 10, + Timeout: 10, + Attempts: 10, + RetryDelay: 10, + } + + opts.Merge(&Options{ + Pipeline: "pipeline", + Delay: 2, + Timeout: 1, + Attempts: 1, + RetryDelay: 1, + }) + + assert.Equal(t, "default", opts.Pipeline) + assert.Equal(t, 10, opts.Attempts) + assert.Equal(t, 10, opts.Delay) + assert.Equal(t, 10, opts.Timeout) + assert.Equal(t, 10, opts.RetryDelay) +} diff --git a/plugins/jobs/structs/job_test.go b/plugins/jobs/structs/job_test.go new file mode 100644 index 00000000..e7240c6b --- /dev/null +++ b/plugins/jobs/structs/job_test.go @@ -0,0 +1,19 @@ +package structs + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestJob_Body(t *testing.T) { + j := &Job{Payload: "hello"} + + assert.Equal(t, []byte("hello"), j.Body()) +} + +func TestJob_Context(t *testing.T) { + j := &Job{Job: "job"} + + assert.Equal(t, []byte(`{"id":"id","job":"job"}`), j.Context("id")) +} diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go index 47d37cc2..0f737fbd 100644 --- a/plugins/kv/drivers/boltdb/driver.go +++ b/plugins/kv/drivers/boltdb/driver.go @@ -9,8 +9,8 @@ import ( "time" "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" "github.com/spiral/roadrunner/v2/utils" diff --git a/plugins/kv/drivers/boltdb/plugin.go b/plugins/kv/drivers/boltdb/plugin.go index 6ae1a1f6..c839130f 100644 --- a/plugins/kv/drivers/boltdb/plugin.go +++ b/plugins/kv/drivers/boltdb/plugin.go @@ -2,12 +2,15 @@ package boltdb import ( "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" ) -const PluginName = "boltdb" +const ( + PluginName string = "boltdb" + RootPluginName string = "kv" +) // Plugin BoltDB K/V storage. type Plugin struct { @@ -21,7 +24,7 @@ type Plugin struct { } func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error { - if !cfg.Has(kv.PluginName) { + if !cfg.Has(RootPluginName) { return errors.E(errors.Disabled) } diff --git a/plugins/kv/drivers/memcached/driver.go b/plugins/kv/drivers/memcached/driver.go index 14e7c078..42e342ac 100644 --- a/plugins/kv/drivers/memcached/driver.go +++ b/plugins/kv/drivers/memcached/driver.go @@ -6,8 +6,8 @@ import ( "github.com/bradfitz/gomemcache/memcache" "github.com/spiral/errors" + kv "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" ) diff --git a/plugins/kv/drivers/memcached/plugin.go b/plugins/kv/drivers/memcached/plugin.go index 22ea5cca..59a2b7cb 100644 --- a/plugins/kv/drivers/memcached/plugin.go +++ b/plugins/kv/drivers/memcached/plugin.go @@ -2,12 +2,15 @@ package memcached import ( "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" ) -const PluginName = "memcached" +const ( + PluginName string = "memcached" + RootPluginName string = "kv" +) type Plugin struct { // config plugin @@ -17,7 +20,7 @@ type Plugin struct { } func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error { - if !cfg.Has(kv.PluginName) { + if !cfg.Has(RootPluginName) { return errors.E(errors.Disabled) } diff --git a/plugins/kv/interface.go b/plugins/kv/interface.go deleted file mode 100644 index 5736a6a7..00000000 --- a/plugins/kv/interface.go +++ /dev/null @@ -1,39 +0,0 @@ -package kv - -import kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" - -// Storage represents single abstract storage. -type Storage interface { - // Has checks if value exists. - Has(keys ...string) (map[string]bool, error) - - // Get loads value content into a byte slice. - Get(key string) ([]byte, error) - - // MGet loads content of multiple values - // Returns the map with existing keys and associated values - MGet(keys ...string) (map[string][]byte, error) - - // Set used to upload item to KV with TTL - // 0 value in TTL means no TTL - Set(items ...*kvv1.Item) error - - // MExpire sets the TTL for multiply keys - MExpire(items ...*kvv1.Item) error - - // TTL return the rest time to live for provided keys - // Not supported for the memcached - TTL(keys ...string) (map[string]string, error) - - // Clear clean the entire storage - Clear() error - - // Delete one or multiple keys. - Delete(keys ...string) error -} - -// Constructor provides storage based on the config -type Constructor interface { - // KVConstruct provides Storage based on the config key - KVConstruct(key string) (Storage, error) -} diff --git a/plugins/kv/plugin.go b/plugins/kv/plugin.go index 03dbaed6..e9ea25df 100644 --- a/plugins/kv/plugin.go +++ b/plugins/kv/plugin.go @@ -5,10 +5,12 @@ import ( endure "github.com/spiral/endure/pkg/container" "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" "github.com/spiral/roadrunner/v2/plugins/logger" ) +// PluginName linked to the memory, boltdb, memcached, redis plugins. DO NOT change w/o sync. const PluginName string = "kv" const ( @@ -25,9 +27,9 @@ const ( type Plugin struct { log logger.Logger // constructors contains general storage constructors, such as boltdb, memory, memcached, redis. - constructors map[string]Constructor + constructors map[string]kv.Constructor // storages contains user-defined storages, such as boltdb-north, memcached-us and so on. - storages map[string]Storage + storages map[string]kv.Storage // KV configuration cfg Config cfgPlugin config.Configurer @@ -43,8 +45,8 @@ func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error { if err != nil { return errors.E(op, err) } - p.constructors = make(map[string]Constructor, 5) - p.storages = make(map[string]Storage, 5) + p.constructors = make(map[string]kv.Constructor, 5) + p.storages = make(map[string]kv.Storage, 5) p.log = log p.cfgPlugin = cfg return nil @@ -203,7 +205,7 @@ func (p *Plugin) Collects() []interface{} { } } -func (p *Plugin) GetAllStorageDrivers(name endure.Named, constructor Constructor) { +func (p *Plugin) GetAllStorageDrivers(name endure.Named, constructor kv.Constructor) { // save the storage constructor p.constructors[name.Name()] = constructor } diff --git a/plugins/kv/rpc.go b/plugins/kv/rpc.go index 3f7ba97c..ad4aefa9 100644 --- a/plugins/kv/rpc.go +++ b/plugins/kv/rpc.go @@ -2,6 +2,7 @@ package kv import ( "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/logger" kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" ) @@ -9,7 +10,7 @@ import ( // Wrapper for the plugin type rpc struct { // all available storages - storages map[string]Storage + storages map[string]kv.Storage // svc is a plugin implementing Storage interface srv *Plugin // Logger diff --git a/plugins/memory/kv.go b/plugins/memory/kv.go index c13c2314..3cec1f97 100644 --- a/plugins/memory/kv.go +++ b/plugins/memory/kv.go @@ -6,8 +6,8 @@ import ( "time" "github.com/spiral/errors" + kv "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" ) diff --git a/plugins/memory/plugin.go b/plugins/memory/plugin.go index 70badf15..7d418a70 100644 --- a/plugins/memory/plugin.go +++ b/plugins/memory/plugin.go @@ -2,9 +2,9 @@ package memory import ( "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/kv" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" ) @@ -21,7 +21,6 @@ type Plugin struct { func (p *Plugin) Init(log logger.Logger, cfg config.Configurer) error { p.log = log - p.log = log p.cfgPlugin = cfg p.stop = make(chan struct{}, 1) return nil diff --git a/plugins/memory/pubsub.go b/plugins/memory/pubsub.go index d027a8a5..3c909900 100644 --- a/plugins/memory/pubsub.go +++ b/plugins/memory/pubsub.go @@ -3,8 +3,8 @@ package memory import ( "sync" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/pkg/bst" - "github.com/spiral/roadrunner/v2/pkg/pubsub" "github.com/spiral/roadrunner/v2/plugins/logger" ) diff --git a/plugins/redis/channel.go b/plugins/redis/channel.go index 5817853c..0cd62d19 100644 --- a/plugins/redis/channel.go +++ b/plugins/redis/channel.go @@ -6,7 +6,7 @@ import ( "github.com/go-redis/redis/v8" "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/logger" "github.com/spiral/roadrunner/v2/utils" ) diff --git a/plugins/redis/kv.go b/plugins/redis/kv.go index 2e4b9bfd..5bf03af1 100644 --- a/plugins/redis/kv.go +++ b/plugins/redis/kv.go @@ -7,8 +7,8 @@ import ( "github.com/go-redis/redis/v8" "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/kv" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" kvv1 "github.com/spiral/roadrunner/v2/proto/kv/v1beta" "github.com/spiral/roadrunner/v2/utils" diff --git a/plugins/redis/plugin.go b/plugins/redis/plugin.go index 9d98790b..3c62a63f 100644 --- a/plugins/redis/plugin.go +++ b/plugins/redis/plugin.go @@ -5,9 +5,9 @@ import ( "github.com/go-redis/redis/v8" "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/kv" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" ) diff --git a/plugins/redis/pubsub.go b/plugins/redis/pubsub.go index 4e41acb5..8bd78514 100644 --- a/plugins/redis/pubsub.go +++ b/plugins/redis/pubsub.go @@ -6,7 +6,7 @@ import ( "github.com/go-redis/redis/v8" "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/config" "github.com/spiral/roadrunner/v2/plugins/logger" ) diff --git a/plugins/server/interface.go b/plugins/server/interface.go index 0424d52d..b0f84a7f 100644 --- a/plugins/server/interface.go +++ b/plugins/server/interface.go @@ -19,5 +19,5 @@ type Server interface { // NewWorker return a new worker with provided and attached by the user listeners and environment variables NewWorker(ctx context.Context, env Env, listeners ...events.Listener) (*worker.Process, error) // NewWorkerPool return new pool of workers (PHP) with attached events listeners, env variables and based on the provided configuration - NewWorkerPool(ctx context.Context, opt pool.Config, env Env, listeners ...events.Listener) (pool.Pool, error) + NewWorkerPool(ctx context.Context, opt *pool.Config, env Env, listeners ...events.Listener) (pool.Pool, error) } diff --git a/plugins/server/plugin.go b/plugins/server/plugin.go index 00639f43..038d83d4 100644 --- a/plugins/server/plugin.go +++ b/plugins/server/plugin.go @@ -21,14 +21,14 @@ import ( "github.com/spiral/roadrunner/v2/utils" ) -// PluginName for the server -const PluginName = "server" - -// RrRelay env variable key (internal) -const RrRelay = "RR_RELAY" - -// RrRPC env variable key (internal) if the RPC presents -const RrRPC = "RR_RPC" +const ( + // PluginName for the server + PluginName = "server" + // RrRelay env variable key (internal) + RrRelay = "RR_RELAY" + // RrRPC env variable key (internal) if the RPC presents + RrRPC = "RR_RPC" +) // Plugin manages worker type Plugin struct { @@ -124,7 +124,7 @@ func (server *Plugin) NewWorker(ctx context.Context, env Env, listeners ...event const op = errors.Op("server_plugin_new_worker") list := make([]events.Listener, 0, len(listeners)) - list = append(list, server.collectWorkerLogs) + list = append(list, server.collectWorkerEvents) spawnCmd, err := server.CmdFactory(env) if err != nil { @@ -140,15 +140,16 @@ func (server *Plugin) NewWorker(ctx context.Context, env Env, listeners ...event } // NewWorkerPool issues new worker pool. -func (server *Plugin) NewWorkerPool(ctx context.Context, opt pool.Config, env Env, listeners ...events.Listener) (pool.Pool, error) { +func (server *Plugin) NewWorkerPool(ctx context.Context, opt *pool.Config, env Env, listeners ...events.Listener) (pool.Pool, error) { const op = errors.Op("server_plugin_new_worker_pool") + spawnCmd, err := server.CmdFactory(env) if err != nil { return nil, errors.E(op, err) } - list := make([]events.Listener, 0, 1) - list = append(list, server.collectEvents) + list := make([]events.Listener, 0, 2) + list = append(list, server.collectPoolEvents, server.collectWorkerEvents) if len(listeners) != 0 { list = append(list, listeners...) } @@ -209,7 +210,7 @@ func (server *Plugin) setEnv(e Env) []string { return env } -func (server *Plugin) collectEvents(event interface{}) { +func (server *Plugin) collectPoolEvents(event interface{}) { if we, ok := event.(events.PoolEvent); ok { switch we.Event { case events.EventMaxMemory: @@ -238,7 +239,9 @@ func (server *Plugin) collectEvents(event interface{}) { server.log.Warn("requested pool restart") } } +} +func (server *Plugin) collectWorkerEvents(event interface{}) { if we, ok := event.(events.WorkerEvent); ok { switch we.Event { case events.EventWorkerError: @@ -264,16 +267,13 @@ func (server *Plugin) collectEvents(event interface{}) { } } -func (server *Plugin) collectWorkerLogs(event interface{}) { - if we, ok := event.(events.WorkerEvent); ok { - switch we.Event { - case events.EventWorkerError: - server.log.Error(strings.TrimRight(we.Payload.(error).Error(), " \n\t")) - case events.EventWorkerLog: - server.log.Debug(strings.TrimRight(utils.AsString(we.Payload.([]byte)), " \n\t")) - // stderr event is INFO level - case events.EventWorkerStderr: - server.log.Info(strings.TrimRight(utils.AsString(we.Payload.([]byte)), " \n\t")) +func (server *Plugin) collectJobsEvents(event interface{}) { //nolint:unused + if jev, ok := event.(events.JobEvent); ok { + switch jev.Event { + case events.EventJobStart: + server.log.Info("Job started", "start", jev.Start, "elapsed", jev.Elapsed) + case events.EventJobOK: + server.log.Info("Job OK", "start", jev.Start, "elapsed", jev.Elapsed) } } } diff --git a/plugins/websockets/executor/executor.go b/plugins/websockets/executor/executor.go index 664b4dfd..c1f79a78 100644 --- a/plugins/websockets/executor/executor.go +++ b/plugins/websockets/executor/executor.go @@ -7,7 +7,7 @@ import ( json "github.com/json-iterator/go" "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/logger" "github.com/spiral/roadrunner/v2/plugins/websockets/commands" "github.com/spiral/roadrunner/v2/plugins/websockets/connection" diff --git a/plugins/websockets/plugin.go b/plugins/websockets/plugin.go index ca5f2f59..5925a588 100644 --- a/plugins/websockets/plugin.go +++ b/plugins/websockets/plugin.go @@ -10,10 +10,10 @@ import ( "github.com/google/uuid" json "github.com/json-iterator/go" "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/pkg/payload" phpPool "github.com/spiral/roadrunner/v2/pkg/pool" "github.com/spiral/roadrunner/v2/pkg/process" - "github.com/spiral/roadrunner/v2/pkg/pubsub" "github.com/spiral/roadrunner/v2/pkg/worker" "github.com/spiral/roadrunner/v2/plugins/broadcast" "github.com/spiral/roadrunner/v2/plugins/config" @@ -106,7 +106,7 @@ func (p *Plugin) Serve() chan error { p.Lock() defer p.Unlock() - p.phpPool, err = p.server.NewWorkerPool(context.Background(), phpPool.Config{ + p.phpPool, err = p.server.NewWorkerPool(context.Background(), &phpPool.Config{ Debug: p.cfg.Pool.Debug, NumWorkers: p.cfg.Pool.NumWorkers, MaxJobs: p.cfg.Pool.MaxJobs, @@ -273,7 +273,7 @@ func (p *Plugin) Reset() error { p.phpPool = nil var err error - p.phpPool, err = p.server.NewWorkerPool(context.Background(), phpPool.Config{ + p.phpPool, err = p.server.NewWorkerPool(context.Background(), &phpPool.Config{ Debug: p.cfg.Pool.Debug, NumWorkers: p.cfg.Pool.NumWorkers, MaxJobs: p.cfg.Pool.MaxJobs, diff --git a/plugins/websockets/pool/workers_pool.go b/plugins/websockets/pool/workers_pool.go index 752ba3ce..758620f6 100644 --- a/plugins/websockets/pool/workers_pool.go +++ b/plugins/websockets/pool/workers_pool.go @@ -4,7 +4,7 @@ import ( "sync" json "github.com/json-iterator/go" - "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/common/pubsub" "github.com/spiral/roadrunner/v2/plugins/logger" "github.com/spiral/roadrunner/v2/plugins/websockets/connection" "github.com/spiral/roadrunner/v2/utils" |