feature: interceptors plugins support

aggregatedpool/workers.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/temporalio/roadrunner-temporal/v4/common"
 	"github.com/temporalio/roadrunner-temporal/v4/internal"
 	tActivity "go.temporal.io/sdk/activity"
 	temporalClient "go.temporal.io/sdk/client"
@@ -15,7 +16,7 @@ import (
 
 const tq = "taskqueue"
 
-func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client) ([]worker.Worker, error) {
+func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client, interceptors map[string]common.Interceptor) ([]worker.Worker, error) {
 	workers := make([]worker.Worker, 0, 1)
 
 	for i := 0; i < len(wi); i++ {
@@ -37,6 +38,9 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo
 
 		// interceptor used here to  the headers
 		wi[i].Options.Interceptors = append(wi[i].Options.Interceptors, NewWorkerInterceptor())
+		for _, interceptor := range interceptors {
+			wi[i].Options.Interceptors = append(wi[i].Options.Interceptors, interceptor.WorkerInterceptor())
+		}
 
 		wrk := worker.New(tc, wi[i].TaskQueue, wi[i].Options)
 

common/interfaces.go

@@ -10,9 +10,15 @@ import (
 	"github.com/roadrunner-server/sdk/v4/state/process"
 	"github.com/roadrunner-server/sdk/v4/worker"
 	"github.com/temporalio/roadrunner-temporal/v4/internal"
+	"go.temporal.io/sdk/interceptor"
 	"go.uber.org/zap"
 )
 
+type Interceptor interface {
+	WorkerInterceptor() interceptor.WorkerInterceptor
+	Name() string
+}
+
 type Pool interface {
 	// Workers returns worker list associated with the pool.
 	Workers() (workers []*worker.Process)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/goccy/go-json v0.10.0
 	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.3.0
+	github.com/roadrunner-server/endure/v2 v2.0.1
 	github.com/roadrunner-server/errors v1.2.0
 	github.com/roadrunner-server/sdk/v4 v4.0.0
 	github.com/stretchr/testify v1.8.1

go.sum

@@ -610,6 +610,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/procfs v0.9.0 h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI=
 github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY=
+github.com/roadrunner-server/endure/v2 v2.0.1 h1:2greoQ669iCjsYNt14dwXXQGHAKDmLtFzL0OIE0ZATc=
+github.com/roadrunner-server/endure/v2 v2.0.1/go.mod h1:RDrC9SFlyCGqGA2v9SqFIA+EqWTFmPxafIb4SMeHCHM=
 github.com/roadrunner-server/errors v1.2.0 h1:qBmNXt8Iex9QnYTjCkbJKsBZu2EtYkQCM06GUDcQBbI=
 github.com/roadrunner-server/errors v1.2.0/go.mod h1:z0ECxZp/dDa5RahtMcy4mBIavVxiZ9vwE5kByl7kFtY=
 github.com/roadrunner-server/goridge/v3 v3.6.2 h1:LH5HXfCygDp05KnOaXpa4fqVPWTsH7V3lfvPtMwFU3k=

plugin.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	prom "github.com/prometheus/client_golang/prometheus"
+	"github.com/roadrunner-server/endure/v2/dep"
 	"github.com/roadrunner-server/errors"
 	"github.com/roadrunner-server/sdk/v4/events"
 	"github.com/roadrunner-server/sdk/v4/metrics"
@@ -89,6 +90,8 @@ type Plugin struct {
 	stopCh   chan struct{}
 
 	workers []worker.Worker
+
+	interceptors map[string]common.Interceptor
 }
 
 func (p *Plugin) Init(cfg common.Configurer, log Logger, server common.Server) error {
@@ -224,6 +227,8 @@ func (p *Plugin) Init(cfg common.Configurer, log Logger, server common.Server) e
 		}
 	}
 
+	p.interceptors = make(map[string]common.Interceptor)
+
 	return nil
 }
 
@@ -428,7 +433,7 @@ func (p *Plugin) Reset() error {
 	}
 
 	// based on the worker info -> initialize workers
-	p.workers, err = aggregatedpool.TemporalWorkers(p.rrWorkflowDef, p.rrActivityDef, wi, p.log, p.client)
+	p.workers, err = aggregatedpool.TemporalWorkers(p.rrWorkflowDef, p.rrActivityDef, wi, p.log, p.client, p.interceptors)
 	if err != nil {
 		return err
 	}
@@ -510,7 +515,7 @@ func (p *Plugin) initPool() error {
 		return err
 	}
 
-	p.workers, err = aggregatedpool.TemporalWorkers(p.rrWorkflowDef, p.rrActivityDef, wi, p.log, p.client)
+	p.workers, err = aggregatedpool.TemporalWorkers(p.rrWorkflowDef, p.rrActivityDef, wi, p.log, p.client, p.interceptors)
 	if err != nil {
 		return err
 	}
@@ -536,3 +541,16 @@ func (p *Plugin) initPool() error {
 
 	return nil
 }
+
+// Collects collecting grpc interceptors
+func (p *Plugin) Collects() []*dep.In {
+	return []*dep.In{
+		dep.Fits(func(pp any) {
+			mdw := pp.(common.Interceptor)
+			// just to be safe
+			p.mu.Lock()
+			p.interceptors[mdw.Name()] = mdw
+			p.mu.Unlock()
+		}, (*common.Interceptor)(nil)),
+	}
+}