go-train-delay-listen.go

go-train-delay-listen.go

func (l *Lister) ListenSubscription(cm chan<- *pubsub.Message, subID string) error {

	defer l.pubsub.Close()

	sub := l.pubsub.Subscription(subID)

	// Receive blocks until the context is cancelled or an error occurs.
	err := sub.Receive(l.ctx, func(ctx context.Context, msg *pubsub.Message) {
		log.Debugf("[ListenSubscription] Receive message from pubsub:%q\n", string(msg.Data))
		cm <- msg
	})
	if err != nil {
		return fmt.Errorf("Receive: %v", err)
	}

	return nil
}