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 {
		
		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
	}

	
	cctx, cancel := context.WithCancel(ctx)

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

	
	
	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 ( " " , zap . String ( " " , options . Project ) , zap . String ( " " , options . Subscription ) , ) if options . Project == " " { return nil , fmt . Errorf ( " " ) } 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 ( " " , 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.
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 .