Documentation
¶
Index ¶
- type Config
- type Options
- type PubSubMQFlow
- func (r *PubSubMQFlow) Characteristics() pipeline.Characteristics
- func (r *PubSubMQFlow) HealthCheck(ctx context.Context) error
- func (r *PubSubMQFlow) QueueBacklog(ctx context.Context) ([]pipeline.QueueBacklogStat, error)
- func (r *PubSubMQFlow) RequestChannels() []pipeline.RequestChannel
- func (r *PubSubMQFlow) ResultChannel() chan api.ResultMessage
- func (r *PubSubMQFlow) RetryChannel() chan pipeline.RetryMessage
- func (r *PubSubMQFlow) Shutdown()
- func (r *PubSubMQFlow) Start(ctx context.Context)
- func (r *PubSubMQFlow) StopConsuming()
- type RequestChannelData
- type TopicConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶ added in v0.9.1
type Config struct {
ProjectID string `json:"project_id"`
ResultTopicID string `json:"result_topic_id"`
BatchSize int `json:"batch_size,omitempty"`
Topics []TopicConfig `json:"topics"`
}
Config is the transport config for the GCP PubSub flow. It is parsed from JSON provided via --transport-config or --transport-config-file.
func LoadConfig ¶ added in v0.9.1
LoadConfig parses, applies defaults, and validates a GCP PubSub Config.
func (*Config) ApplyDefaults ¶ added in v0.9.1
func (c *Config) ApplyDefaults()
type Options ¶
type Options struct {
IGWBaseURL string
ProjectID string
RequestPathURL string
InferenceObjective string
RequestSubscriberID string
ResultTopicID string
TopicsConfigFile string
BatchSize int
}
Options holds CLI flags for the GCP PubSub flow.
func NewOptions ¶
func NewOptions() *Options
func (*Options) HasTopicConfig ¶
HasTopicConfig reports whether a multi-topic configuration file is set.
type PubSubMQFlow ¶
type PubSubMQFlow struct {
// contains filtered or unexported fields
}
func NewGCPPubSubMQFlow ¶
func NewGCPPubSubMQFlow(cfg Config, workerPools []pipeline.WorkerPoolConfig, gateFactory pipeline.GateFactory) (*PubSubMQFlow, error)
NewGCPPubSubMQFlow builds a GCP Pub/Sub flow from a parsed Config. The config is expected to have had ApplyDefaults applied (LoadConfig does this). workerPools resolves the named pool each topic routes to; gateFactory, when non-nil, instantiates a per-topic gate for any topic that declares a gate_type.
func (*PubSubMQFlow) Characteristics ¶
func (r *PubSubMQFlow) Characteristics() pipeline.Characteristics
func (*PubSubMQFlow) HealthCheck ¶
func (r *PubSubMQFlow) HealthCheck(ctx context.Context) error
HealthCheck backs the /readyz probe by confirming each configured request subscription is reachable. It prefers the passive signal from the consume loop (requestWorker): a recent successful Receive proves the broker round-tripped without any extra API call or IAM permission, and a recent Receive error marks the pod not-ready. When the consume loop has no fresh signal (a quiet queue, or before Start), it falls back to an active GetSubscription probe. An unreachable Pub/Sub backend surfaces as a gRPC Unavailable error and marks the pod not-ready, fixing the original bug where such pods reported ready.
func (*PubSubMQFlow) QueueBacklog ¶
func (r *PubSubMQFlow) QueueBacklog(ctx context.Context) ([]pipeline.QueueBacklogStat, error)
QueueBacklog reports the number of undelivered messages per subscription, sourced from the Cloud Monitoring metric pubsub.googleapis.com/subscription/num_undelivered_messages. The value is approximate and lags real time by the metric's sampling interval.
ExpiringCounts is intentionally never set: Cloud Monitoring exposes only the aggregate undelivered-message count, not per-item deadlines, so the view stays nil and the poller emits nothing for it.
func (*PubSubMQFlow) RequestChannels ¶
func (r *PubSubMQFlow) RequestChannels() []pipeline.RequestChannel
func (*PubSubMQFlow) ResultChannel ¶
func (r *PubSubMQFlow) ResultChannel() chan api.ResultMessage
func (*PubSubMQFlow) RetryChannel ¶
func (r *PubSubMQFlow) RetryChannel() chan pipeline.RetryMessage
func (*PubSubMQFlow) Shutdown ¶
func (r *PubSubMQFlow) Shutdown()
func (*PubSubMQFlow) Start ¶
func (r *PubSubMQFlow) Start(ctx context.Context)
func (*PubSubMQFlow) StopConsuming ¶
func (r *PubSubMQFlow) StopConsuming()
type RequestChannelData ¶
type RequestChannelData struct {
// contains filtered or unexported fields
}
type TopicConfig ¶
type TopicConfig struct {
SubscriberID string `json:"subscriber_id"`
WorkerPoolID string `json:"worker_pool_id"`
InferenceObjective string `json:"inference_objective"`
RequestPathURL string `json:"request_path_url"`
IGWBaseURL string `json:"igw_base_url"`
pipeline.GateConfig
Labels map[string]string `json:"labels,omitempty"`
}
Source Files