summaryrefslogtreecommitdiff
path: root/plugins/temporal/activity
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/temporal/activity')
-rw-r--r--plugins/temporal/activity/activity_pool.go197
-rw-r--r--plugins/temporal/activity/plugin.go215
-rw-r--r--plugins/temporal/activity/rpc.go66
3 files changed, 0 insertions, 478 deletions
diff --git a/plugins/temporal/activity/activity_pool.go b/plugins/temporal/activity/activity_pool.go
deleted file mode 100644
index d09722ce..00000000
--- a/plugins/temporal/activity/activity_pool.go
+++ /dev/null
@@ -1,197 +0,0 @@
-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"
-)
-
-// RR_MODE env variable
-const RR_MODE = "RR_MODE" //nolint:golint,stylecheck
-// RR_CODEC env variable
-const RR_CODEC = "RR_CODEC" //nolint:golint,stylecheck
-
-//
-const doNotCompleteOnReturn = "doNotCompleteOnReturn"
-
-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)
-}
-
-type 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) {
- const op = errors.Op("new_activity_pool")
- // env variables
- env := map[string]string{RR_MODE: RRMode, RR_CODEC: codec.GetName()}
- wp, err := server.NewWorkerPool(context.Background(), poolConfig, env, listener)
- if err != nil {
- return nil, errors.E(op, 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 {
- const op = errors.Op("activity_pool_start")
- pool.dc = temporal.GetDataConverter()
-
- err := pool.initWorkers(ctx, temporal)
- if err != nil {
- return errors.E(op, err)
- }
-
- for i := 0; i < len(pool.tWorkers); i++ {
- err := pool.tWorkers[i].Start()
- if err != nil {
- return errors.E(op, 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) {
- const op = errors.Op("activity_pool_get_activity_context")
- c, ok := pool.running.Load(string(taskToken))
- if !ok {
- return nil, errors.E(op, errors.Str("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("activity_pool_create_temporal_worker")
-
- 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 i := 0; i < len(workerInfo); i++ {
- w, err := temporal.CreateWorker(workerInfo[i].TaskQueue, workerInfo[i].Options)
- if err != nil {
- return errors.E(op, err, pool.Destroy(ctx))
- }
-
- pool.tWorkers = append(pool.tWorkers, w)
- for j := 0; j < len(workerInfo[i].Activities); j++ {
- w.RegisterActivityWithOptions(pool.executeActivity, activity.RegisterOptions{
- Name: workerInfo[i].Activities[j].Name,
- DisableAlreadyRegisteredCheck: false,
- })
-
- pool.activities = append(pool.activities, workerInfo[i].Activities[j].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("activity_pool_execute_activity")
-
- 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)
- var 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, errors.Str("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
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
-}
diff --git a/plugins/temporal/activity/rpc.go b/plugins/temporal/activity/rpc.go
deleted file mode 100644
index 49efcd4f..00000000
--- a/plugins/temporal/activity/rpc.go
+++ /dev/null
@@ -1,66 +0,0 @@
-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
-}
generated by cgit v1.2.3 (git 2.25.1) at 2025-02-02 07:50:08 +0000