func()

in graphql/handler/transport/http_multipart_mixed.go [41:152]

• 72 lines of code
• 9 McCabe index (conditional complexity)

func (t MultipartMixed) Do(w http.ResponseWriter, r *http.Request, exec graphql.GraphExecutor) {
	// Implements the multipart/mixed spec as a multipart/mixed response:
	// * https://github.com/graphql/graphql-wg/blob/e4ef5f9d5997815d9de6681655c152b6b7838b4c/rfcs/DeferStream.md
	//   2022/08/23 as implemented by gqlgen.
	// * https://github.com/graphql/graphql-wg/blob/f22ea7748c6ebdf88fdbf770a8d9e41984ebd429/rfcs/DeferStream.md June 2023 Spec for the
	//   `incremental` field
	// * https://github.com/graphql/graphql-over-http/blob/main/rfcs/IncrementalDelivery.md
	//   multipart specification
	// Follows the format that is used in the Apollo Client tests:
	// https://github.com/apollographql/apollo-client/blob/v3.11.8/src/link/http/__tests__/responseIterator.ts#L68
	// Apollo Client, despite mentioning in its requests that they require the 2022 spec, it wants the
	// `incremental` field to be an array of responses, not a single response. Theoretically we could
	// batch responses in the `incremental` field, if we wanted to optimize this code.
	ctx := r.Context()
	flusher, ok := w.(http.Flusher)
	if !ok {
		SendErrorf(w, http.StatusInternalServerError, "streaming unsupported")
		return
	}
	defer flusher.Flush()

	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	// This header will be replaced below, but it's required in case we return errors.
	w.Header().Set("Content-Type", "application/json")

	boundary := t.Boundary
	if boundary == "" {
		boundary = "-"
	}
	timeout := t.DeliveryTimeout
	if timeout.Milliseconds() < 1 {
		// If the timeout is less than 1ms, we'll set it to 1ms to avoid a busy loop
		timeout = 1 * time.Millisecond
	}

	params := &graphql.RawParams{}
	start := graphql.Now()
	params.Headers = r.Header
	params.ReadTime = graphql.TraceTiming{
		Start: start,
		End:   graphql.Now(),
	}

	bodyString, err := getRequestBody(r)
	if err != nil {
		gqlErr := gqlerror.Errorf("could not get json request body: %+v", err)
		resp := exec.DispatchError(ctx, gqlerror.List{gqlErr})
		log.Printf("could not get json request body: %+v", err.Error())
		writeJson(w, resp)
		return
	}

	bodyReader := io.NopCloser(strings.NewReader(bodyString))
	if err = jsonDecode(bodyReader, &params); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		gqlErr := gqlerror.Errorf(
			"json request body could not be decoded: %+v body:%s",
			err,
			bodyString,
		)
		resp := exec.DispatchError(ctx, gqlerror.List{gqlErr})
		log.Printf("decoding error: %+v body:%s", err.Error(), bodyString)
		writeJson(w, resp)
		return
	}

	rc, opErr := exec.CreateOperationContext(ctx, params)
	ctx = graphql.WithOperationContext(ctx, rc)
	if opErr != nil {
		w.WriteHeader(statusFor(opErr))

		resp := exec.DispatchError(ctx, opErr)
		writeJson(w, resp)
		return
	}

	// Example of the response format (note the new lines and boundaries are important!):
	// https://github.com/graphql/graphql-over-http/blob/main/rfcs/IncrementalDelivery.md
	// --graphql
	// Content-Type: application/json
	//
	// {"data":{"apps":{"apps":[ .. ],"totalNumApps":161,"__typename":"AppsOutput"}},"hasNext":true}
	// --graphql
	// Content-Type: application/json
	//
	// {"incremental":[{"data":{"groupAccessCount":0},"label":"test","path":["apps","apps",7],"hasNext":true}],"hasNext":true}
	// --graphql
	// ...
	// --graphql--
	// Last boundary is a closing boundary with two dashes at the end.

	w.Header().Set(
		"Content-Type",
		fmt.Sprintf(`multipart/mixed;boundary="%s";deferSpec=20220824`, boundary),
	)

	a := newMultipartResponseAggregator(w, boundary, timeout)
	defer a.Done(w)

	responses, ctx := exec.DispatchOperation(ctx, rc)
	initialResponse := true
	for {
		response := responses(ctx)
		if response == nil {
			break
		}

		a.Add(response, initialResponse)
		initialResponse = false
	}
}