1 files changed, 215 insertions, 0 deletions
diff --git a/plugins/temporal/activity/plugin.go b/plugins/temporal/activity/plugin.go
new file mode 100644
index 00000000..5e562a8d
--- /dev/null
+++ b/plugins/temporal/activity/plugin.go
@@ -0,0 +1,215 @@
+package activity
+
+import (
+	"context"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/spiral/roadrunner/v2/pkg/events"
+	"github.com/spiral/roadrunner/v2/pkg/worker"
+
+	"sync"
+	"sync/atomic"
+
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+	"github.com/spiral/roadrunner/v2/plugins/server"
+	"github.com/spiral/roadrunner/v2/plugins/temporal/client"
+)
+
+const (
+	// PluginName defines public service name.
+	PluginName = "activities"
+
+	// RRMode sets as RR_MODE env variable to let worker know about the mode to run.
+	RRMode = "temporal/activity"
+)
+
+// Plugin to manage activity execution.
+type Plugin struct {
+	temporal client.Temporal
+	events   events.Handler
+	server   server.Server
+	log      logger.Logger
+	mu       sync.Mutex
+	reset    chan struct{}
+	pool     activityPool
+	closing  int64
+}
+
+// Init configures activity service.
+func (p *Plugin) Init(temporal client.Temporal, server server.Server, log logger.Logger) error {
+	const op = errors.Op("activity_plugin_init")
+	if temporal.GetConfig().Activities == nil {
+		// no need to serve activities
+		return errors.E(op, errors.Disabled)
+	}
+
+	p.temporal = temporal
+	p.server = server
+	p.events = events.NewEventsHandler()
+	p.log = log
+	p.reset = make(chan struct{})
+
+	return nil
+}
+
+// Serve activities with underlying workers.
+func (p *Plugin) Serve() chan error {
+	const op = errors.Op("activity_plugin_serve")
+
+	errCh := make(chan error, 1)
+	pool, err := p.startPool()
+	if err != nil {
+		errCh <- errors.E(op, err)
+		return errCh
+	}
+
+	p.pool = pool
+
+	go func() {
+		for {
+			select {
+			case <-p.reset:
+				if atomic.LoadInt64(&p.closing) == 1 {
+					return
+				}
+
+				err := p.replacePool()
+				if err == nil {
+					continue
+				}
+
+				bkoff := backoff.NewExponentialBackOff()
+				bkoff.InitialInterval = time.Second
+
+				err = backoff.Retry(p.replacePool, bkoff)
+				if err != nil {
+					errCh <- errors.E(op, err)
+				}
+			}
+		}
+	}()
+
+	return errCh
+}
+
+// Stop stops the serving plugin.
+func (p *Plugin) Stop() error {
+	atomic.StoreInt64(&p.closing, 1)
+	const op = errors.Op("activity_plugin_stop")
+
+	pool := p.getPool()
+	if pool != nil {
+		p.pool = nil
+		err := pool.Destroy(context.Background())
+		if err != nil {
+			return errors.E(op, err)
+		}
+		return nil
+	}
+
+	return nil
+}
+
+// Name of the service.
+func (p *Plugin) Name() string {
+	return PluginName
+}
+
+// RPC returns associated rpc service.
+func (p *Plugin) RPC() interface{} {
+	return &rpc{srv: p, client: p.temporal.GetClient()}
+}
+
+// Workers returns pool workers.
+func (p *Plugin) Workers() []worker.SyncWorker {
+	return p.getPool().Workers()
+}
+
+// ActivityNames returns list of all available activities.
+func (p *Plugin) ActivityNames() []string {
+	return p.pool.ActivityNames()
+}
+
+// Reset resets underlying workflow pool with new copy.
+func (p *Plugin) Reset() error {
+	p.reset <- struct{}{}
+
+	return nil
+}
+
+// AddListener adds event listeners to the service.
+func (p *Plugin) AddListener(listener events.Listener) {
+	p.events.AddListener(listener)
+}
+
+// AddListener adds event listeners to the service.
+func (p *Plugin) poolListener(event interface{}) {
+	if ev, ok := event.(events.PoolEvent); ok {
+		if ev.Event == events.EventPoolError {
+			p.log.Error("Activity pool error", "error", ev.Payload.(error))
+			p.reset <- struct{}{}
+		}
+	}
+
+	p.events.Push(event)
+}
+
+// AddListener adds event listeners to the service.
+func (p *Plugin) startPool() (activityPool, error) {
+	pool, err := newActivityPool(
+		p.temporal.GetCodec().WithLogger(p.log),
+		p.poolListener,
+		*p.temporal.GetConfig().Activities,
+		p.server,
+	)
+
+	if err != nil {
+		return nil, errors.E(errors.Op("newActivityPool"), err)
+	}
+
+	err = pool.Start(context.Background(), p.temporal)
+	if err != nil {
+		return nil, errors.E(errors.Op("startActivityPool"), err)
+	}
+
+	p.log.Debug("Started activity processing", "activities", pool.ActivityNames())
+
+	return pool, nil
+}
+
+func (p *Plugin) replacePool() error {
+	pool, err := p.startPool()
+	if err != nil {
+		p.log.Error("Replace activity pool failed", "error", err)
+		return errors.E(errors.Op("newActivityPool"), err)
+	}
+
+	p.log.Debug("Replace activity pool")
+
+	var previous activityPool
+
+	p.mu.Lock()
+	previous, p.pool = p.pool, pool
+	p.mu.Unlock()
+
+	errD := previous.Destroy(context.Background())
+	if errD != nil {
+		p.log.Error(
+			"Unable to destroy expired activity pool",
+			"error",
+			errors.E(errors.Op("destroyActivityPool"), errD),
+		)
+	}
+
+	return nil
+}
+
+// getPool returns currently pool.
+func (p *Plugin) getPool() activityPool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	return p.pool
+}