|
| 1 | +// Copyright 2026 Google LLC |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +// Package pubsub provides a sublauncher that adds PubSub trigger capabilities to ADK web server. |
| 16 | +package pubsub |
| 17 | + |
| 18 | +import ( |
| 19 | + "flag" |
| 20 | + "fmt" |
| 21 | + "net/http" |
| 22 | + "strings" |
| 23 | + "time" |
| 24 | + |
| 25 | + "github.com/gorilla/mux" |
| 26 | + |
| 27 | + "google.golang.org/adk/cmd/launcher" |
| 28 | + "google.golang.org/adk/cmd/launcher/web" |
| 29 | + "google.golang.org/adk/internal/cli/util" |
| 30 | + "google.golang.org/adk/server/adkrest/controllers/triggers" |
| 31 | +) |
| 32 | + |
| 33 | +type pubsubConfig struct { |
| 34 | + pathPrefix string |
| 35 | + triggerMaxRetries int |
| 36 | + triggerBaseDelay time.Duration |
| 37 | + triggerMaxDelay time.Duration |
| 38 | + triggerMaxRuns int |
| 39 | +} |
| 40 | + |
| 41 | +type pubsubLauncher struct { |
| 42 | + flags *flag.FlagSet |
| 43 | + config *pubsubConfig |
| 44 | +} |
| 45 | + |
| 46 | +// NewLauncher creates a new pubsub launcher. It extends Web launcher. |
| 47 | +func NewLauncher() web.Sublauncher { |
| 48 | + config := &pubsubConfig{} |
| 49 | + |
| 50 | + fs := flag.NewFlagSet("pubsub", flag.ContinueOnError) |
| 51 | + fs.StringVar(&config.pathPrefix, "path_prefix", "/api", "Path prefix for the PubSub trigger endpoint. Default is '/api'.") |
| 52 | + fs.IntVar(&config.triggerMaxRetries, "trigger_max_retries", 3, "Maximum retries for HTTP 429 errors from triggers") |
| 53 | + fs.DurationVar(&config.triggerBaseDelay, "trigger_base_delay", 1*time.Second, "Base delay for trigger retry exponential backoff") |
| 54 | + fs.DurationVar(&config.triggerMaxDelay, "trigger_max_delay", 10*time.Second, "Maximum delay for trigger retry exponential backoff") |
| 55 | + fs.IntVar(&config.triggerMaxRuns, "trigger_max_concurrent_runs", 100, "Maximum concurrent trigger runs") |
| 56 | + |
| 57 | + return &pubsubLauncher{ |
| 58 | + config: config, |
| 59 | + flags: fs, |
| 60 | + } |
| 61 | +} |
| 62 | + |
| 63 | +// Keyword implements web.Sublauncher. Returns the command-line keyword for pubsub launcher. |
| 64 | +func (p *pubsubLauncher) Keyword() string { |
| 65 | + return "pubsub" |
| 66 | +} |
| 67 | + |
| 68 | +// Parse parses the command-line arguments for the pubsub launcher. |
| 69 | +func (p *pubsubLauncher) Parse(args []string) ([]string, error) { |
| 70 | + err := p.flags.Parse(args) |
| 71 | + if err != nil || !p.flags.Parsed() { |
| 72 | + return nil, fmt.Errorf("failed to parse pubsub flags: %v", err) |
| 73 | + } |
| 74 | + if p.config.triggerMaxRetries < 0 { |
| 75 | + return nil, fmt.Errorf("trigger_max_retries must be >= 0") |
| 76 | + } |
| 77 | + if p.config.triggerBaseDelay < 0 { |
| 78 | + return nil, fmt.Errorf("trigger_base_delay must be >= 0") |
| 79 | + } |
| 80 | + if p.config.triggerMaxDelay < 0 { |
| 81 | + return nil, fmt.Errorf("trigger_max_delay must be >= 0") |
| 82 | + } |
| 83 | + if p.config.triggerMaxRuns < 0 { |
| 84 | + return nil, fmt.Errorf("trigger_max_concurrent_runs must be >= 0") |
| 85 | + } |
| 86 | + |
| 87 | + prefix := p.config.pathPrefix |
| 88 | + if !strings.HasPrefix(prefix, "/") { |
| 89 | + prefix = "/" + prefix |
| 90 | + } |
| 91 | + p.config.pathPrefix = strings.TrimSuffix(prefix, "/") |
| 92 | + |
| 93 | + return p.flags.Args(), nil |
| 94 | +} |
| 95 | + |
| 96 | +// CommandLineSyntax returns the command-line syntax for the pubsub launcher. |
| 97 | +func (p *pubsubLauncher) CommandLineSyntax() string { |
| 98 | + return util.FormatFlagUsage(p.flags) |
| 99 | +} |
| 100 | + |
| 101 | +// SimpleDescription implements web.Sublauncher. |
| 102 | +func (p *pubsubLauncher) SimpleDescription() string { |
| 103 | + return "starts ADK PubSub trigger endpoint server" |
| 104 | +} |
| 105 | + |
| 106 | +// SetupSubrouters adds the PubSub trigger endpoint to the parent router. |
| 107 | +func (p *pubsubLauncher) SetupSubrouters(router *mux.Router, config *launcher.Config) error { |
| 108 | + triggerConfig := triggers.TriggerConfig{ |
| 109 | + MaxRetries: p.config.triggerMaxRetries, |
| 110 | + BaseDelay: p.config.triggerBaseDelay, |
| 111 | + MaxDelay: p.config.triggerMaxDelay, |
| 112 | + MaxConcurrentRuns: p.config.triggerMaxRuns, |
| 113 | + } |
| 114 | + |
| 115 | + controller := triggers.NewPubSubController( |
| 116 | + config.SessionService, |
| 117 | + config.AgentLoader, |
| 118 | + config.MemoryService, |
| 119 | + config.ArtifactService, |
| 120 | + config.PluginConfig, |
| 121 | + triggerConfig, |
| 122 | + ) |
| 123 | + |
| 124 | + subrouter := router |
| 125 | + if p.config.pathPrefix != "" && p.config.pathPrefix != "/" { |
| 126 | + subrouter = router.PathPrefix(p.config.pathPrefix).Subrouter() |
| 127 | + } |
| 128 | + |
| 129 | + subrouter.HandleFunc("/apps/{app_name}/trigger/pubsub", controller.PubSubTriggerHandler).Methods(http.MethodPost) |
| 130 | + return nil |
| 131 | +} |
| 132 | + |
| 133 | +// UserMessage implements web.Sublauncher. |
| 134 | +func (p *pubsubLauncher) UserMessage(webURL string, printer func(v ...any)) { |
| 135 | + printer(fmt.Sprintf(" pubsub: PubSub trigger endpoint is available at %s%s/apps/{app_name}/trigger/pubsub", webURL, p.config.pathPrefix)) |
| 136 | +} |
0 commit comments