Expand defer tests

client/incremental_http.go

@@ -0,0 +1,195 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"mime"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+)
+
+type IncrementalHandler struct {
+	close func() error
+	next  func(response any) error
+}
+
+func (i *IncrementalHandler) Close() error {
+	return i.close()
+}
+
+func (i *IncrementalHandler) Next(response any) error {
+	return i.next(response)
+}
+
+type IncrementalInitialResponse struct {
+	Data       any             `json:"data"`
+	Label      string          `json:"label"`
+	Path       []any           `json:"path"`
+	HasNext    bool            `json:"hasNext"`
+	Errors     json.RawMessage `json:"errors"`
+	Extensions map[string]any  `json:"extensions"`
+}
+
+type IncrementalData struct {
+	// Support for "items" for @stream is not yet available, only "data" for
+	// @defer, as per the 2023 spec. Similarly, this retains a more complete
+	// list of fields, but not "id," and represents a mid-point between the
+	// 2022 and 2023 specs.
+
+	Data       any             `json:"data"`
+	Label      string          `json:"label"`
+	Path       []any           `json:"path"`
+	HasNext    bool            `json:"hasNext"`
+	Errors     json.RawMessage `json:"errors"`
+	Extensions map[string]any  `json:"extensions"`
+}
+
+type IncrementalResponse struct {
+	// Does not include the pending or completed fields from the 2023 spec.
+
+	Incremental []IncrementalData `json:"incremental"`
+	HasNext     bool              `json:"hasNext"`
+	Errors      json.RawMessage   `json:"errors"`
+	Extensions  map[string]any    `json:"extensions"`
+}
+
+func errorIncremental(err error) *IncrementalHandler {
+	return &IncrementalHandler{
+		close: func() error { return nil },
+		next: func(response any) error {
+			return err
+		},
+	}
+}
+
+// IncrementalHTTP returns a GraphQL response handler for the current
+// GQLGen implementation of the [incremental delivery over HTTP spec].
+// This spec provides for "streaming" responses triggered by the use of
+// @stream or @defer using is an alternate approach to SSE. To that end,
+// the client retains the interface of the handler returned from
+// Client.SSE.
+//
+// IncrementalHTTP delivery using multipart/mixed is just the structure
+// of the response: the payloads are specified by the defer-stream spec,
+// which are in transition. For more detail, see the links in the
+// definition for transport.MultipartMixed. We use the name
+// IncrementalHTTP here to distinguish from the multipart form upload
+// (the term "multipart" usually referring to the latter).
+//
+// IncrementalHandler is not safe for concurrent use, or for production
+// use at all.
+//
+// [incremental delivery over HTTP spec]: https://github.com/graphql/graphql-over-http/blob/main/rfcs/IncrementalDelivery.md
+func (p *Client) IncrementalHTTP(ctx context.Context, query string, options ...Option) *IncrementalHandler {
+	r, err := p.newRequest(query, options...)
+	if err != nil {
+		return errorIncremental(fmt.Errorf("request: %w", err))
+	}
+	r.Header.Set("Accept", "multipart/mixed")
+
+	w := httptest.NewRecorder()
+	p.h.ServeHTTP(w, r)
+
+	res := w.Result() //nolint:bodyclose // Remains open since we are reading from it incrementally.
+	if res.StatusCode >= http.StatusBadRequest {
+		return errorIncremental(fmt.Errorf("http %d: %s", w.Code, w.Body.String()))
+	}
+	mediaType, params, err := mime.ParseMediaType(res.Header.Get("Content-Type"))
+	if err != nil {
+		return errorIncremental(fmt.Errorf("parse content-type: %w", err))
+	}
+	if mediaType != "multipart/mixed" {
+		return errorIncremental(fmt.Errorf("expected content-type multipart/mixed, got %s", mediaType))
+	}
+
+	// TODO: worth checking the deferSpec either to confirm this client
+	// supports it exactly, or simply to make sure it is within some
+	// expected range.
+	deferSpec, ok := params["deferspec"]
+	if !ok || deferSpec == "" {
+		return errorIncremental(errors.New("expected deferSpec in content-type"))
+	}
+
+	boundary, ok := params["boundary"]
+	if !ok || boundary == "" {
+		return errorIncremental(errors.New("expected boundary in content-type"))
+	}
+	mr := multipart.NewReader(res.Body, boundary)
+
+	ctx, cancel := context.WithCancelCause(ctx)
+	initial := true
+
+	return &IncrementalHandler{
+		close: func() error {
+			res.Body.Close()
+			cancel(context.Canceled)
+			return nil
+		},
+		next: func(response any) (err error) {
+			defer func() {
+				if err != nil {
+					res.Body.Close()
+					cancel(err)
+				}
+			}()
+
+			var data any
+			var rawErrors json.RawMessage
+
+			type nextPart struct {
+				*multipart.Part
+				Err error
+			}
+
+			nextPartCh := make(chan nextPart)
+			go func() {
+				var next nextPart
+				next.Part, next.Err = mr.NextPart()
+				nextPartCh <- next
+			}()
+
+			var next nextPart
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case next = <-nextPartCh:
+			}
+
+			if next.Err == io.EOF {
+				res.Body.Close()
+				cancel(context.Canceled)
+				return nil
+			}
+			if err = next.Err; err != nil {
+				return err
+			}
+			if ct := next.Header.Get("Content-Type"); ct != "application/json" {
+				err = fmt.Errorf(`expected content-type "application/json", got %q`, ct)
+				return err
+			}
+
+			if initial {
+				initial = false
+				data = IncrementalInitialResponse{}
+			} else {
+				data = IncrementalResponse{}
+			}
+			if err = json.NewDecoder(next.Part).Decode(&data); err != nil {
+				return err
+			}
+
+			// We want to unpack even if there is an error, so we can see partial
+			// responses.
+			err = unpack(data, response, p.dc)
+			if len(rawErrors) != 0 {
+				err = RawJsonError{rawErrors}
+				return err
+			}
+			return err
+		},
+	}
+}

codegen/testserver/followschema/defer.graphql

@@ -1,6 +1,6 @@
 extend type Query {
-    deferCase1: DeferModel
-    deferCase2: [DeferModel!]
+    deferSingle: DeferModel
+    deferMultiple: [DeferModel!]
 }
 
 type DeferModel {

codegen/testserver/followschema/resolver.go

@@ -197,13 +197,13 @@ func (r *queryResolver) DefaultParameters(ctx context.Context, falsyBoolean *boo
 	panic("not implemented")
 }
 
-// DeferCase1 is the resolver for the deferCase1 field.
-func (r *queryResolver) DeferCase1(ctx context.Context) (*DeferModel, error) {
+// DeferSingle is the resolver for the deferSingle field.
+func (r *queryResolver) DeferSingle(ctx context.Context) (*DeferModel, error) {
 	panic("not implemented")
 }
 
-// DeferCase2 is the resolver for the deferCase2 field.
-func (r *queryResolver) DeferCase2(ctx context.Context) ([]*DeferModel, error) {
+// DeferMultiple is the resolver for the deferMultiple field.
+func (r *queryResolver) DeferMultiple(ctx context.Context) ([]*DeferModel, error) {
 	panic("not implemented")
 }