Initial commit of the Temporal plugins set

3 files changed, 475 insertions, 0 deletions

diff --git a/plugins/temporal/activity/activity_pool.go b/plugins/temporal/activity/activity_pool.go
new file mode 100644
index 00000000..0aa2a62f
--- /dev/null
+++ b/plugins/temporal/activity/activity_pool.go
@@ -0,0 +1,199 @@
+package activity
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/pkg/events"
+	"github.com/spiral/roadrunner/v2/pkg/pool"
+	rrWorker "github.com/spiral/roadrunner/v2/pkg/worker"
+	"github.com/spiral/roadrunner/v2/plugins/server"
+	"github.com/spiral/roadrunner/v2/plugins/temporal/client"
+	rrt "github.com/spiral/roadrunner/v2/plugins/temporal/protocol"
+	"go.temporal.io/api/common/v1"
+	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/converter"
+	"go.temporal.io/sdk/internalbindings"
+	"go.temporal.io/sdk/worker"
+)
+
+type (
+	activityPool interface {
+		Start(ctx context.Context, temporal client.Temporal) error
+		Destroy(ctx context.Context) error
+		Workers() []rrWorker.SyncWorker
+		ActivityNames() []string
+		GetActivityContext(taskToken []byte) (context.Context, error)
+	}
+
+	activityPoolImpl struct {
+		dc         converter.DataConverter
+		codec      rrt.Codec
+		seqID      uint64
+		activities []string
+		wp         pool.Pool
+		tWorkers   []worker.Worker
+		running    sync.Map
+	}
+)
+
+// newActivityPool
+func newActivityPool(
+	codec rrt.Codec,
+	listener events.Listener,
+	poolConfig pool.Config,
+	server server.Server,
+) (activityPool, error) {
+	wp, err := server.NewWorkerPool(
+		context.Background(),
+		poolConfig,
+		map[string]string{"RR_MODE": RRMode, "RR_CODEC": codec.GetName()},
+		listener,
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &activityPoolImpl{
+		codec:   codec,
+		wp:      wp,
+		running: sync.Map{},
+	}, nil
+}
+
+// initWorkers request workers info from underlying PHP and configures temporal workers linked to the pool.
+func (pool *activityPoolImpl) Start(ctx context.Context, temporal client.Temporal) error {
+	pool.dc = temporal.GetDataConverter()
+
+	err := pool.initWorkers(ctx, temporal)
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < len(pool.tWorkers); i++ {
+		err := pool.tWorkers[i].Start()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// initWorkers request workers info from underlying PHP and configures temporal workers linked to the pool.
+func (pool *activityPoolImpl) Destroy(ctx context.Context) error {
+	for i := 0; i < len(pool.tWorkers); i++ {
+		pool.tWorkers[i].Stop()
+	}
+
+	pool.wp.Destroy(ctx)
+	return nil
+}
+
+// Workers returns list of all allocated workers.
+func (pool *activityPoolImpl) Workers() []rrWorker.SyncWorker {
+	return pool.wp.Workers()
+}
+
+// ActivityNames returns list of all available activity names.
+func (pool *activityPoolImpl) ActivityNames() []string {
+	return pool.activities
+}
+
+// ActivityNames returns list of all available activity names.
+func (pool *activityPoolImpl) GetActivityContext(taskToken []byte) (context.Context, error) {
+	c, ok := pool.running.Load(string(taskToken))
+	if !ok {
+		return nil, errors.E("heartbeat on non running activity")
+	}
+
+	return c.(context.Context), nil
+}
+
+// initWorkers request workers workflows from underlying PHP and configures temporal workers linked to the pool.
+func (pool *activityPoolImpl) initWorkers(ctx context.Context, temporal client.Temporal) error {
+	const op = errors.Op("createTemporalWorker")
+
+	workerInfo, err := rrt.FetchWorkerInfo(pool.codec, pool.wp, temporal.GetDataConverter())
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	pool.activities = make([]string, 0)
+	pool.tWorkers = make([]worker.Worker, 0)
+
+	for _, info := range workerInfo {
+		w, err := temporal.CreateWorker(info.TaskQueue, info.Options)
+		if err != nil {
+			return errors.E(op, err, pool.Destroy(ctx))
+		}
+
+		pool.tWorkers = append(pool.tWorkers, w)
+		for _, activityInfo := range info.Activities {
+			w.RegisterActivityWithOptions(pool.executeActivity, activity.RegisterOptions{
+				Name:                          activityInfo.Name,
+				DisableAlreadyRegisteredCheck: false,
+			})
+
+			pool.activities = append(pool.activities, activityInfo.Name)
+		}
+	}
+
+	return nil
+}
+
+// executes activity with underlying worker.
+func (pool *activityPoolImpl) executeActivity(ctx context.Context, args *common.Payloads) (*common.Payloads, error) {
+	const op = errors.Op("executeActivity")
+
+	heartbeatDetails := &common.Payloads{}
+	if activity.HasHeartbeatDetails(ctx) {
+		err := activity.GetHeartbeatDetails(ctx, &heartbeatDetails)
+		if err != nil {
+			return nil, errors.E(op, err)
+		}
+	}
+
+	var (
+		info = activity.GetInfo(ctx)
+		msg  = rrt.Message{
+			ID: atomic.AddUint64(&pool.seqID, 1),
+			Command: rrt.InvokeActivity{
+				Name:             info.ActivityType.Name,
+				Info:             info,
+				HeartbeatDetails: len(heartbeatDetails.Payloads),
+			},
+			Payloads: args,
+		}
+	)
+
+	if len(heartbeatDetails.Payloads) != 0 {
+		msg.Payloads.Payloads = append(msg.Payloads.Payloads, heartbeatDetails.Payloads...)
+	}
+
+	pool.running.Store(string(info.TaskToken), ctx)
+	defer pool.running.Delete(string(info.TaskToken))
+
+	result, err := pool.codec.Execute(pool.wp, rrt.Context{TaskQueue: info.TaskQueue}, msg)
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	if len(result) != 1 {
+		return nil, errors.E(op, "invalid activity worker response")
+	}
+
+	out := result[0]
+	if out.Failure != nil {
+		if out.Failure.Message == "doNotCompleteOnReturn" {
+			return nil, activity.ErrResultPending
+		}
+
+		return nil, internalbindings.ConvertFailureToError(out.Failure, pool.dc)
+	}
+
+	return out.Payloads, nil
+}
diff --git a/plugins/temporal/activity/plugin.go b/plugins/temporal/activity/plugin.go
new file mode 100644
index 00000000..02d66297
--- /dev/null
+++ b/plugins/temporal/activity/plugin.go
@@ -0,0 +1,210 @@
+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 {
+	if temporal.GetConfig().Activities == nil {
+		// no need to serve activities
+		return errors.E(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 {
+	errCh := make(chan error, 1)
+
+	pool, err := p.startPool()
+	if err != nil {
+		errCh <- errors.E("startPool", 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("deadPool", err)
+				}
+			}
+		}
+	}()
+
+	return errCh
+}
+
+// Stop stops the serving plugin.
+func (p *Plugin) Stop() error {
+	atomic.StoreInt64(&p.closing, 1)
+
+	pool := p.getPool()
+	if pool != nil {
+		p.pool = nil
+		return pool.Destroy(context.Background())
+	}
+
+	return nil
+}
+
+// Name of the service.
+func (p *Plugin) Name() string {
+	return PluginName
+}
+
+// RPC returns associated rpc service.
+func (p *Plugin) RPC() interface{} {
+	client, _ := p.temporal.GetClient()
+
+	return &rpc{srv: p, client: client}
+}
+
+// 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
+}
diff --git a/plugins/temporal/activity/rpc.go b/plugins/temporal/activity/rpc.go
new file mode 100644
index 00000000..49efcd4f
--- /dev/null
+++ b/plugins/temporal/activity/rpc.go
@@ -0,0 +1,66 @@
+package activity
+
+import (
+	v1Proto "github.com/golang/protobuf/proto" //nolint:staticcheck
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/client"
+	"google.golang.org/protobuf/proto"
+)
+
+/*
+- the method's type is exported.
+- the method is exported.
+- the method has two arguments, both exported (or builtin) types.
+- the method's second argument is a pointer.
+- the method has return type error.
+*/
+type rpc struct {
+	srv    *Plugin
+	client client.Client
+}
+
+// RecordHeartbeatRequest sent by activity to record current state.
+type RecordHeartbeatRequest struct {
+	TaskToken []byte `json:"taskToken"`
+	Details   []byte `json:"details"`
+}
+
+// RecordHeartbeatResponse sent back to the worker to indicate that activity was cancelled.
+type RecordHeartbeatResponse struct {
+	Canceled bool `json:"canceled"`
+}
+
+// RecordActivityHeartbeat records heartbeat for an activity.
+// taskToken - is the value of the binary "TaskToken" field of the "ActivityInfo" struct retrieved inside the activity.
+// details - is the progress you want to record along with heart beat for this activity.
+// The errors it can return:
+// - EntityNotExistsError
+// - InternalServiceError
+// - CanceledError
+func (r *rpc) RecordActivityHeartbeat(in RecordHeartbeatRequest, out *RecordHeartbeatResponse) error {
+	details := &commonpb.Payloads{}
+
+	if len(in.Details) != 0 {
+		if err := proto.Unmarshal(in.Details, v1Proto.MessageV2(details)); err != nil {
+			return err
+		}
+	}
+
+	// find running activity
+	ctx, err := r.srv.getPool().GetActivityContext(in.TaskToken)
+	if err != nil {
+		return err
+	}
+
+	activity.RecordHeartbeat(ctx, details)
+
+	select {
+	case <-ctx.Done():
+		*out = RecordHeartbeatResponse{Canceled: true}
+	default:
+		*out = RecordHeartbeatResponse{Canceled: false}
+	}
+
+	return nil
+}