Method subscribeTask

pkg/events/redis/redis.go:160–204

(ctx context.Context)

)

func (ps *PubSub) subscribeTask(ctx context.Context) error {
	logger := log.FromContext(ctx)
	ch := ps.sub.Channel()
	store := &PubSubStore{
		PubSub: ps,
		// NOTE: only for reading; no additional settings needed.
	}
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case msg, ok := <-ch:
			if !ok {
				return errChannelClosed.New()
			}
			var evtPB *ttnpb.Event
			switch {
			case strings.HasPrefix(msg.Payload, protoEncodingPrefix):
				evtPB = &ttnpb.Event{}
				err := decodeEventData(msg.Payload, evtPB)
				if err != nil {
					logger.WithError(err).Warn("Failed to decode event payload")
					continue
				}
			case strings.HasPrefix(msg.Payload, metaEncodingPrefix):
				m := strings.Split(strings.TrimPrefix(msg.Payload, metaEncodingPrefix), " ")
				var err error
				evtPB, err = store.LoadEvent(ctx, m[0])
				if err != nil {
					logger.WithError(err).Warn("Failed to load event payload")
					continue
				}
			default:
				logger.Warn("Skip decoding event with unexpected encoding")
				continue
			}
			evt, err := events.FromProto(evtPB)
			if err != nil {
				logger.WithError(err).Warn("Failed to convert event from protobuf")
				continue
			}
			ps.PubSub.Publish(&patternEvent{Event: evt, pattern: msg.Pattern})
		}
	}
}

type patternEvent struct {
	events.Event