roadrunner_temporal

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 official plugin of RoadRunner and available out of the box in RoadRunner 2.0.

Read more about application server installation here.

To install PHP-SDK:

$ composer require temporal/sdk

Read how to configure your worker and init workflows here.

Testing

To test integration make sure to install Golang and PHP 7.4+ at the same host. Composer is required to manage the extension.

$ make install-dependencies
$ make start-docker-compose
$ make test

License

MIT License

Documentation

Index

• Constants
• type ClientAuthType
• type Config
  • func (c *Config) InitDefault() error
• type Metrics
• type Plugin
  • func (p *Plugin) Init(cfg common.Configurer, log *zap.Logger, server common.Server) error
  • func (p *Plugin) MetricsCollector() []prom.Collector
  • func (p *Plugin) Name() string
  • func (p *Plugin) RPC() any
  • func (p *Plugin) Reset() error
  • func (p *Plugin) ResetAP() error
  • func (p *Plugin) SedID() uint64
  • func (p *Plugin) Serve() chan error
  • func (p *Plugin) Stop() error
  • func (p *Plugin) Workers() []*process.State
• type Prometheus
• type RecordHeartbeatRequest
• type RecordHeartbeatResponse
• type Statsd
• type StatsdReporterConfig
• type TLS

Constants

const (
	// PluginName defines public service name.
	PluginName string = "temporal"

	// 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"
)

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 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) Init

func (p *Plugin) Init(cfg common.Configurer, log *zap.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) Reset

func (p *Plugin) Reset() error

func (*Plugin) ResetAP

func (p *Plugin) ResetAP() error

func (*Plugin) SedID

func (p *Plugin) SedID() uint64

func (*Plugin) Serve

func (p *Plugin) Serve() chan error

func (*Plugin) Stop

func (p *Plugin) Stop() 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
}