rrtemporal

README

Roadrunner Temporal

The repository contains a number of plugins which enable workflow and activity processing for PHP processes. The communication protocol, supervisor, load-balancer is based on RoadRunner PHP Application Server.

Installation

Temporal is an official plugin of RoadRunner and available out-of-the-box in >= RoadRunner 2023.0.

Read more about application server installation here.

To install PHP-SDK:

$ composer require temporal/sdk

License

MIT License

Documentation

Index

• Constants
• func ActivitiesInfo(wi []*internal.WorkerInfo) map[string]*internal.ActivityInfo
• func WorkerInfo(c common.Codec, p common.Pool, rrVersion string) ([]*internal.WorkerInfo, error)
• func WorkflowsInfo(wi []*internal.WorkerInfo) map[string]*internal.WorkflowInfo
• type ClientAuthType
• type Config
  • func (c *Config) InitDefault() error
• type Informer
• type Logger
• type Metrics
• type Plugin
  • func (p *Plugin) AddWorker() error
  • func (p *Plugin) Collects() []*dep.In
  • func (p *Plugin) Init(cfg common.Configurer, log Logger, server common.Server) error
  • func (p *Plugin) MetricsCollector() []prom.Collector
  • func (p *Plugin) Name() string
  • func (p *Plugin) RPC() any
  • func (p *Plugin) Ready() (*status.Status, error)
  • func (p *Plugin) RemoveWorker(ctx context.Context) error
  • func (p *Plugin) Reset() error
  • func (p *Plugin) ResetAP() error
  • func (p *Plugin) Serve() chan error
  • func (p *Plugin) Status() (*status.Status, error)
  • func (p *Plugin) Stop(context.Context) error
  • func (p *Plugin) Workers() []*process.State
• type Prometheus
• type RecordHeartbeatRequest
• type RecordHeartbeatResponse
• type Statsd
• type StatsdReporterConfig
• type TLS

Constants

const (

	// RrMode env variable key
	RrMode string = "RR_MODE"
	// RrCodec env variable key
	RrCodec string = "RR_CODEC"
	// RrCodecVal - codec name, should be in sync with the PHP-SDK
	RrCodecVal string = "protobuf"
)

const (
	MetricsTypeSummary string = "summary"
)

Functions

func ActivitiesInfo

func ActivitiesInfo(wi []*internal.WorkerInfo) map[string]*internal.ActivityInfo

func WorkerInfo

func WorkerInfo(c common.Codec, p common.Pool, rrVersion string) ([]*internal.WorkerInfo, error)

func WorkflowsInfo

func WorkflowsInfo(wi []*internal.WorkerInfo) map[string]*internal.WorkflowInfo

Types

type ClientAuthType

type ClientAuthType string

const (
	NoClientCert               ClientAuthType = "no_client_cert"
	RequestClientCert          ClientAuthType = "request_client_cert"
	RequireAnyClientCert       ClientAuthType = "require_any_client_cert"
	VerifyClientCertIfGiven    ClientAuthType = "verify_client_cert_if_given"
	RequireAndVerifyClientCert ClientAuthType = "require_and_verify_client_cert"
)

type Config

type Config struct {
	Metrics    *Metrics     `mapstructure:"metrics"`
	Activities *pool.Config `mapstructure:"activities"`
	TLS        *TLS         `mapstructure:"tls, omitempty"`

	Address   string `mapstructure:"address"`
	Namespace string `mapstructure:"namespace"`
	CacheSize int    `mapstructure:"cache_size"`
}

Config of the temporal client and dependent services.

func (*Config) InitDefault

func (c *Config) InitDefault() error

type Informer

type Informer interface {
	Workers() []*process.State
}

Informer used to get workers from particular plugin or set of plugins

type Logger

type Logger interface {
	NamedLogger(name string) *zap.Logger
}

type Metrics

type Metrics struct {
	Driver     string      `mapstructure:"driver"`
	Prometheus *Prometheus `mapstructure:"prometheus"`
	Statsd     *Statsd     `mapstructure:"statsd"`
}

type Plugin

type Plugin struct {
	// contains filtered or unexported fields
}

func (*Plugin) AddWorker

func (p *Plugin) AddWorker() error

func (*Plugin) Collects

func (p *Plugin) Collects() []*dep.In

Collects collecting grpc interceptors

func (*Plugin) Init

func (p *Plugin) Init(cfg common.Configurer, log Logger, server common.Server) error

func (*Plugin) MetricsCollector

func (p *Plugin) MetricsCollector() []prom.Collector

func (*Plugin) Name

func (p *Plugin) Name() string

func (*Plugin) RPC

func (p *Plugin) RPC() any

func (*Plugin) Ready

func (p *Plugin) Ready() (*status.Status, error)

Ready return readiness status of the particular plugin

func (*Plugin) RemoveWorker

func (p *Plugin) RemoveWorker(ctx context.Context) error

func (*Plugin) Reset

func (p *Plugin) Reset() error

func (*Plugin) ResetAP

func (p *Plugin) ResetAP() error

func (*Plugin) Serve

func (p *Plugin) Serve() chan error

func (*Plugin) Status

func (p *Plugin) Status() (*status.Status, error)

Status return status of the particular plugin

func (*Plugin) Stop

func (p *Plugin) Stop(context.Context) error

func (*Plugin) Workers

func (p *Plugin) Workers() []*process.State

type Prometheus

type Prometheus struct {
	Address string `mapstructure:"address"`
	Type    string `mapstructure:"type"`
	Prefix  string `mapstructure:"prefix"`
}

type RecordHeartbeatRequest

type RecordHeartbeatRequest struct {
	TaskToken []byte `json:"taskToken"`
	Details   []byte `json:"details"`
}

RecordHeartbeatRequest sent by activity to record current state.

type RecordHeartbeatResponse

type RecordHeartbeatResponse struct {
	Canceled bool `json:"canceled"`
}

RecordHeartbeatResponse sent back to the worker to indicate that activity was canceled.

type Statsd

type Statsd struct {
	// The host and port of the statsd server
	HostPort string `mapstructure:"host_port" validate:"nonzero"`
	// The prefix to use in reporting to statsd
	Prefix string `mapstructure:"prefix" validate:"nonzero"`
	// FlushInterval is the maximum interval for sending packets.
	// If it is not specified, it defaults to 1 second.
	FlushInterval time.Duration `mapstructure:"flush_interval"`
	// FlushBytes specifies the maximum udp packet size you wish to send.
	// If FlushBytes is unspecified, it defaults  to 1432 bytes, which is
	// considered safe for local traffic.
	FlushBytes int `mapstructure:"flush_bytes"`
	// Tags to pass to the Tally scope options
	Tags map[string]string `mapstructure:"tags"`
	// TagPrefix ...
	TagPrefix string `mapstructure:"tag_prefix"`
	// TagSeparator allows tags to be appended with a separator. If not specified tag keys and values
	// are embedded to the stat name directly.
	TagSeparator string `mapstructure:"tag_separator"`
}

ref:https://github.dev/temporalio/temporal/common/metrics/config.go:79

type StatsdReporterConfig

type StatsdReporterConfig struct {
	// TagSeparator allows tags to be appended with a separator. If not specified tag keys and values
	// are embedded to the stat name directly.
	TagSeparator string `yaml:"tag_separator"`
}

type TLS

type TLS struct {
	Key        string         `mapstructure:"key"`
	Cert       string         `mapstructure:"cert"`
	RootCA     string         `mapstructure:"root_ca"`
	AuthType   ClientAuthType `mapstructure:"client_auth_type"`
	ServerName string         `mapstructure:"server_name"`
	// contains filtered or unexported fields
}