OpenSubscriptionURL opens a pubsub.Subscription based on u.
(ctx context.Context, u *url.URL)
| 132 | |
| 133 | // OpenSubscriptionURL opens a pubsub.Subscription based on u. |
| 134 | func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error) { |
| 135 | opts := o.SubscriptionOptions |
| 136 | for param, value := range u.Query() { |
| 137 | switch param { |
| 138 | case "prefetch_count": |
| 139 | if len(value) != 1 || len(value[0]) == 0 { |
| 140 | return nil, fmt.Errorf("open subscription %v: invalid query parameter %q", u, param) |
| 141 | } |
| 142 | |
| 143 | prefetchCount, err := strconv.Atoi(value[0]) |
| 144 | if err != nil { |
| 145 | return nil, fmt.Errorf("open subscription %v: invalid query parameter %q: %w", u, param, err) |
| 146 | } |
| 147 | |
| 148 | opts.PrefetchCount = &prefetchCount |
| 149 | default: |
| 150 | return nil, fmt.Errorf("open subscription %v: invalid query parameter %q", u, param) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | queueName := path.Join(u.Host, u.Path) |
| 155 | return OpenSubscription(o.Connection, queueName, &opts), nil |
| 156 | } |
| 157 | |
| 158 | type topic struct { |
| 159 | exchange string // the AMQP exchange |
nothing calls this directly
no test coverage detected