- Rework dispatcher, pipeline, job (not completely)

Create a config sample with RR2 support. Progress on root JOBS plugin.

Signed-off-by: Valery Piashchynski <[email protected]>

1 files changed, 29 insertions, 0 deletions

diff --git a/plugins/jobs/plugin.go b/plugins/jobs/plugin.go
index e708d0ca..42203871 100644
--- a/plugins/jobs/plugin.go
+++ b/plugins/jobs/plugin.go
@@ -1,6 +1,7 @@
 package jobs
 
 import (
+	endure "github.com/spiral/endure/pkg/container"
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
@@ -11,11 +12,25 @@ const (
 )
 
 type Plugin struct {
+	cfg *Config
+	log logger.Logger
+
+	consumers map[string]Consumer
 }
 
 func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) 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)
+	}
 
+	p.consumers = make(map[string]Consumer)
+	p.log = log
 	return nil
 }
 
@@ -29,8 +44,22 @@ func (p *Plugin) Stop() error {
 	return nil
 }
 
+func (p *Plugin) Collects() []interface{} {
+	return []interface{}{
+		p.CollectMQBrokers,
+	}
+}
+
+func (p *Plugin) CollectMQBrokers(name endure.Named, c Consumer) {
+	p.consumers[name.Name()] = c
+}
+
 func (p *Plugin) Available() {}
 
 func (p *Plugin) Name() string {
 	return PluginName
 }
+
+func (p *Plugin) RPC() interface{} {
+	return &rpc{log: p.log}
+}