Method OpenSubscriptionURL

pubsub/gcppubsub/gcppubsub.go:241–280

OpenSubscriptionURL opens a pubsub.Subscription based on u.

(ctx context.Context, u *url.URL)

// OpenSubscriptionURL opens a pubsub.Subscription based on u.
func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error) {
	// Set subscription options to use defaults
	opts := o.SubscriptionOptions

	for param, value := range u.Query() {
		switch param {
		case "max_recv_batch_size":
			maxBatchSize, err := queryParameterInt(value)
			if err != nil {
				return nil, fmt.Errorf("open subscription %v: invalid query parameter %q: %v", u, param, err)
			}

			if maxBatchSize <= 0 || maxBatchSize > 1000 {
				return nil, fmt.Errorf("open subscription %v: invalid query parameter %q: must be between 1 and 1000", u, param)
			}

			opts.MaxBatchSize = maxBatchSize
		case "nacklazy":
			var err error
			nackLazy, err := queryParameterBool(value)
			if err != nil {
				return nil, fmt.Errorf("open subscription %v: invalid query parameter %q: %v", u, param, err)
			}
			opts.NackLazy = nackLazy
		default:
			return nil, fmt.Errorf("open subscription %v: invalid query parameter %q", u, param)
		}
	}
	sc, err := SubscriberClient(ctx, o.Conn)
	if err != nil {
		return nil, err
	}
	subPath := path.Join(u.Host, u.Path)
	if subscriptionPathRE.MatchString(subPath) {
		return OpenSubscriptionByPath(sc, subPath, &opts)
	}
	// Shortened form?
	subName := strings.TrimPrefix(u.Path, "/")
	return OpenSubscription(sc, gcp.ProjectID(u.Host), subName, &opts), nil
}

type topic struct {
	path   string