receiver/pubsub/pubsub.go
func newPubSub(client *pubsub.Client, name string, options *Options, store func(*points.Points)) (*PubSub, error) {
	logger := zapwriter.Logger(name)
	logger.Info("starting google pubsub receiver",
		zap.String("project", options.Project),
		zap.String("subscription", options.Subscription),
	)

	if options.Project == "" {
		return nil, fmt.Errorf("'project' must be specified")
	}

	ctx := context.Background()
	if client == nil {
		c, err := pubsub.NewClient(ctx, options.Project)
		if err != nil {
			return nil, err
		}
		client = c
	}

	sub := client.Subscription(options.Subscription)
	exists, err := sub.Exists(ctx)
	if err != nil {
		return nil, err
	}
	if !exists {
		// TODO: try to create subscription
		return nil, fmt.Errorf("subscription %s in project %s does not exist", options.Subscription, options.Project)
	}

	if options.ReceiverGoRoutines != 0 {
		sub.ReceiveSettings.NumGoroutines = options.ReceiverGoRoutines
	}
	if options.ReceiverMaxBytes != 0 {
		sub.ReceiveSettings.MaxOutstandingBytes = options.ReceiverMaxBytes
	}
	if options.ReceiverMaxMessages != 0 {
		sub.ReceiveSettings.MaxOutstandingMessages = options.ReceiverMaxMessages
	}

	// cancel() will be called to signal the subscription Receive() goroutines to finish and shutdown
	cctx, cancel := context.WithCancel(ctx)

	rcv := &PubSub{
		out:          store,
		name:         name,
		client:       client,
		cancel:       cancel,
		subscription: sub,
		logger:       logger,
		closed:       make(chan struct{}),
	}

	// Receive() will create goroutines as necessary to handle incoming messages. Reconnect
	// on error. Stop processing messages only if Stop() is called
	go func() {
		for {
			err := rcv.subscription.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
				rcv.handleMessage(m)
				m.Ack()
			})
			if err == context.Canceled {
				close(rcv.closed)
				rcv.Stop()
				break
			}
			if err != nil {
				rcv.logger.Error(err.Error())
			}
			time.Sleep(1 * time.Second)
		}
	}()

	return rcv, nil
}
func newPubSub(client *pubsub.Client, name string, options *Options, store func(*points.Points)) (*PubSub, error) {
	logger := zapwriter.Logger(name)
	logger.Info("starting google pubsub receiver",
		zap.String("project", options.Project),
		zap.String("subscription", options.Subscription),
	)

	if options.Project == "" {
		return nil, fmt.Errorf("'project' must be specified")
	}

	ctx := context.Background()
	if client == nil {
		c, err := pubsub.NewClient(ctx, options.Project)
		if err != nil {
			return nil, err
		}
		client = c
	}

	sub := client.Subscription(options.Subscription)
	exists, err := sub.Exists(ctx)
	if err != nil {
		return nil, err
	}
	if !exists {
		// TODO: try to create subscription
		return nil, fmt.Errorf("subscription %s in project %s does not exist", options.Subscription, options.Project)
	}

	if options.ReceiverGoRoutines != 0 {
		sub.ReceiveSettings.NumGoroutines = options.ReceiverGoRoutines
	}
	if options.ReceiverMaxBytes != 0 {
		sub.ReceiveSettings.MaxOutstandingBytes = options.ReceiverMaxBytes
	}
	if options.ReceiverMaxMessages != 0 {
		sub.ReceiveSettings.MaxOutstandingMessages = options.ReceiverMaxMessages
	}

	// cancel() will be called to signal the subscription Receive() goroutines to finish and shutdown
	cctx, cancel := context.WithCancel(ctx)

	rcv := &PubSub{
		out:          store,
		name:         name,
		client:       client,
		cancel:       cancel,
		subscription: sub,
		logger:       logger,
		closed:       make(chan struct{}),
	}

	// Receive() will create goroutines as necessary to handle incoming messages. Reconnect
	// on error. Stop processing messages only if Stop() is called
	go func() {
		for {
			err := rcv.subscription.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
				rcv.handleMessage(m)
				m.Ack()
			})
			if err == context.Canceled {
				close(rcv.closed)
				rcv.Stop()
				break
			}
			if err != nil {
				rcv.logger.Error(err.Error())
			}
			time.Sleep(1 * time.Second)
		}
	}()

	return rcv, nil
}
// newPubSub returns a PubSub receiver. Optionally accepts a client to allow
// for injecting the fake for tests. If client is nil a real
// client connection to the google pubsub service will be attempted.