Method registerSubscriber

subscribe.go:199–287

registerSubscriber initializes the connection.

(ctx context.Context, w http.ResponseWriter, r *http.Request)

// registerSubscriber initializes the connection.
func (h *Hub) registerSubscriber(ctx context.Context, w http.ResponseWriter, r *http.Request) (*LocalSubscriber, *responseController) { //nolint:funlen
	ctx, span := startSpan(ctx, "mercure.subscribe", trace.WithSpanKind(trace.SpanKindConsumer))
	defer span.End()

	s := NewLocalSubscriber(h.retrieveLastEventID(ctx, r), h.logger, h.topicSelectorStore)

	var (
		privateTopics []string
		claims        *claims
	)

	if h.subscriberJWTKeyFunc != nil { //nolint:nestif
		var err error

		claims, err = h.authorize(r, false)
		if claims != nil {
			s.Claims = claims
			privateTopics = claims.Mercure.Subscribe
		}

		if err != nil || (claims == nil && !h.anonymous) {
			http.Error(w, http.StatusText(http.StatusUnauthorized), http.StatusUnauthorized)

			if h.logger.Enabled(ctx, slog.LevelDebug) {
				h.logger.LogAttrs(ctx, slog.LevelDebug, "Subscriber unauthorized", slog.Any("error", err))
			}

			if err != nil {
				recordSpanError(span, err)
			}

			return nil, nil
		}
	}

	topics := r.URL.Query()["topic"]
	if len(topics) == 0 {
		http.Error(w, `Missing "topic" parameter.`, http.StatusBadRequest)

		return nil, nil
	}

	if len(topics) > maxQueryTopics {
		http.Error(w, `Too many "topic" parameters.`, http.StatusBadRequest)

		return nil, nil
	}

	s.SetTopics(topics, privateTopics)

	if span.IsRecording() {
		span.SetAttributes(
			attribute.String("mercure.subscriber.id", s.ID),
			attribute.StringSlice("mercure.topics", topics),
		)
	}

	addCtx := context.WithoutCancel(ctx)