Method subscribeTask

pkg/events/cloud/cloud.go:82–129

(ctx context.Context)

}

func (ps *PubSub) subscribeTask(ctx context.Context) error {
	if ps.subURL == "" {
		return nil
	}
	logger := log.FromContext(ctx)
	subscription, err := pubsub.OpenSubscription(ctx, ps.subURL)
	if err != nil {
		return err
	}
	logger.Info("Subscribed")
	defer func() {
		if err := subscription.Shutdown(ctx); err != nil {
			logger.WithError(err).Warn("Failed to close Go Cloud subscription")
		} else {
			logger.Info("Unsubscribed")
		}
	}()
	for {
		msg, err := subscription.Receive(ctx)
		if err != nil {
			return err
		}
		msg.Ack()
		m := msg.Metadata["content-type"]
		var evt events.Event
		switch m {
		case "application/protobuf":
			var e ttnpb.Event
			if err = proto.Unmarshal(msg.Body, &e); err != nil {
				logger.WithError(err).Warn("Failed to unmarshal event from binary")
				continue
			}
			if evt, err = events.FromProto(&e); err != nil {
				logger.WithError(err).Warn("Failed to unmarshal event from protobuf")
				continue
			}
		case "application/json":
			if evt, err = events.UnmarshalJSON(msg.Body); err != nil {
				logger.WithError(err).Warn("Failed to unmarshal event from JSON")
				continue
			}
		default:
			logger.WithField("content_type", m).Warn("Received event with unknown content type")
			continue
		}
		ps.PubSub.Publish(evt)
	}
}

// Subscribe to events from Go Cloud.
func (ps *PubSub) Subscribe(