1 files changed, 0 insertions, 215 deletions
diff --git a/plugins/temporal/activity/plugin.go b/plugins/temporal/activity/plugin.go
deleted file mode 100644
index 5e562a8d..00000000
--- a/plugins/temporal/activity/plugin.go
+++ /dev/null
@@ -1,215 +0,0 @@
-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
-}