Merge pull request #1 from remorses/subscription-execution

Subscription execution

Author: bhoriuchi

subscription.go

@@ -5,66 +5,128 @@ import (
 	"fmt"
 
 	"github.com/graphql-go/graphql/gqlerrors"
-	"github.com/graphql-go/graphql/language/ast"
+	"github.com/graphql-go/graphql/language/parser"
+	"github.com/graphql-go/graphql/language/source"
 )
 
 // SubscribeParams parameters for subscribing
 type SubscribeParams struct {
-	Schema          Schema
-	Document        *ast.Document
-	RootValue       interface{}
-	ContextValue    context.Context
+	Schema        Schema
+	RequestString string
+	RootValue     interface{}
+	// ContextValue    context.Context
 	VariableValues  map[string]interface{}
 	OperationName   string
 	FieldResolver   FieldResolveFn
 	FieldSubscriber FieldResolveFn
 }
 
-// Subscribe performs a subscribe operation
-func Subscribe(ctx context.Context, p SubscribeParams) chan *Result {
-	resultChannel := make(chan *Result)
+// Subscribe performs a subscribe operation on the given query and schema
+// To finish a subscription you can simply close the channel from inside the `Subscribe` function
+// currently does not support extensions hooks
+func Subscribe(p Params) chan *Result {
+
+	source := source.NewSource(&source.Source{
+		Body: []byte(p.RequestString),
+		Name: "GraphQL request",
+	})
+
+	// TODO run extensions hooks
+
+	// parse the source
+	AST, err := parser.Parse(parser.ParseParams{Source: source})
+	if err != nil {
+
+		// merge the errors from extensions and the original error from parser
+		return sendOneResultAndClose(&Result{
+			Errors: gqlerrors.FormatErrors(err),
+		})
+	}
+
+	// validate document
+	validationResult := ValidateDocument(&p.Schema, AST, nil)
+
+	if !validationResult.IsValid {
+		// run validation finish functions for extensions
+		return sendOneResultAndClose(&Result{
+			Errors: validationResult.Errors,
+		})
+
+	}
+	return ExecuteSubscription(ExecuteParams{
+		Schema:        p.Schema,
+		Root:          p.RootObject,
+		AST:           AST,
+		OperationName: p.OperationName,
+		Args:          p.VariableValues,
+		Context:       p.Context,
+	})
+}
+
+func sendOneResultAndClose(res *Result) chan *Result {
+	resultChannel := make(chan *Result, 1)
+	resultChannel <- res
+	close(resultChannel)
+	return resultChannel
+}
+
+// ExecuteSubscription is similar to graphql.Execute but returns a channel instead of a Result
+// currently does not support extensions
+func ExecuteSubscription(p ExecuteParams) chan *Result {
+
+	if p.Context == nil {
+		p.Context = context.Background()
+	}
 
 	var mapSourceToResponse = func(payload interface{}) *Result {
 		return Execute(ExecuteParams{
 			Schema:        p.Schema,
 			Root:          payload,
-			AST:           p.Document,
+			AST:           p.AST,
 			OperationName: p.OperationName,
-			Args:          p.VariableValues,
-			Context:       p.ContextValue,
+			Args:          p.Args,
+			Context:       p.Context,
 		})
 	}
-
+	var resultChannel = make(chan *Result)
 	go func() {
-		result := &Result{}
+		defer close(resultChannel)
 		defer func() {
 			if err := recover(); err != nil {
-				result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-				resultChannel <- result
+				e, ok := err.(error)
+				if !ok {
+					return
+				}
+				resultChannel <- &Result{
+					Errors: gqlerrors.FormatErrors(e),
+				}
 			}
-			close(resultChannel)
+			return
 		}()
 
 		exeContext, err := buildExecutionContext(buildExecutionCtxParams{
 			Schema:        p.Schema,
-			Root:          p.RootValue,
-			AST:           p.Document,
+			Root:          p.Root,
+			AST:           p.AST,
 			OperationName: p.OperationName,
-			Args:          p.VariableValues,
-			Result:        result,
-			Context:       p.ContextValue,
+			Args:          p.Args,
+			Context:       p.Context,
 		})
 
 		if err != nil {
-			result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-			resultChannel <- result
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(err),
+			}
+
 			return
 		}
 
 		operationType, err := getOperationRootType(p.Schema, exeContext.Operation)
 		if err != nil {
-			result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-			resultChannel <- result
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(err),
+			}
+
 			return
 		}
 
@@ -85,18 +147,20 @@ func Subscribe(ctx context.Context, p SubscribeParams) chan *Result {
 		fieldDef := getFieldDef(p.Schema, operationType, fieldName)
 
 		if fieldDef == nil {
-			err := fmt.Errorf("the subscription field %q is not defined", fieldName)
-			result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-			resultChannel <- result
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(fmt.Errorf("the subscription field %q is not defined", fieldName)),
+			}
+
 			return
 		}
 
-		resolveFn := p.FieldSubscriber
+		resolveFn := fieldDef.Subscribe
+
 		if resolveFn == nil {
-			resolveFn = DefaultResolveFn
-		}
-		if fieldDef.Subscribe != nil {
-			resolveFn = fieldDef.Subscribe
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(fmt.Errorf("the subscription function %q is not defined", fieldName)),
+			}
+			return
 		}
 		fieldPath := &ResponsePath{
 			Key: responseName,
@@ -117,21 +181,24 @@ func Subscribe(ctx context.Context, p SubscribeParams) chan *Result {
 		}
 
 		fieldResult, err := resolveFn(ResolveParams{
-			Source:  p.RootValue,
+			Source:  p.Root,
 			Args:    args,
 			Info:    info,
-			Context: p.ContextValue,
+			Context: p.Context,
 		})
 		if err != nil {
-			result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-			resultChannel <- result
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(err),
+			}
+
 			return
 		}
 
 		if fieldResult == nil {
-			err := fmt.Errorf("no field result")
-			result.Errors = append(result.Errors, gqlerrors.FormatError(err.(error)))
-			resultChannel <- result
+			resultChannel <- &Result{
+				Errors: gqlerrors.FormatErrors(fmt.Errorf("no field result")),
+			}
+
 			return
 		}
 
@@ -140,10 +207,13 @@ func Subscribe(ctx context.Context, p SubscribeParams) chan *Result {
 			sub := fieldResult.(chan interface{})
 			for {
 				select {
-				case <-ctx.Done():
+				case <-p.Context.Done():
 					return
 
-				case res := <-sub:
+				case res, more := <-sub:
+					if !more {
+						return
+					}
 					resultChannel <- mapSourceToResponse(res)
 				}
 			}