Repository: jensneuse/graphql-go-tools
Pull request #136 (2019-12-04 16:14:02)

WIP: add nats datasource


Patch:
diff --git a/go.mod b/go.mod
index 1cb22046..c1a91442 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/gobwas/pool v0.2.0 // indirect
 	github.com/gobwas/ws v1.0.2
 	github.com/jensneuse/diffview v1.0.0
+	github.com/nats-io/nats.go v1.9.1
 	github.com/pkg/errors v0.8.1 // indirect
 	github.com/sebdah/goldie v0.0.0-20180424091453-8784dd1ab561
 	github.com/stretchr/testify v1.4.0 // indirect
diff --git a/go.sum b/go.sum
index 7404f535..73b51c76 100644
--- a/go.sum
+++ b/go.sum
@@ -55,6 +55,14 @@ github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/nats-io/jwt v0.3.0 h1:xdnzwFETV++jNc4W1mw//qFyJGb2ABOombmZJQS4+Qo=
+github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg=
+github.com/nats-io/nats.go v1.9.1 h1:ik3HbLhZ0YABLto7iX80pZLPw/6dx3T+++MZJwLnMrQ=
+github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzEE/Zbp4w=
+github.com/nats-io/nkeys v0.1.0 h1:qMd4+pRHgdr1nAClu+2h/2a5F2TmKcCzjCDazVgRoX4=
+github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
@@ -110,6 +118,8 @@ go.uber.org/zap v1.11.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190621222207-cc06ce4a13d4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
diff --git a/pkg/execution/a_execution-packr.go b/pkg/execution/a_execution-packr.go
index b7961ed7..6c2611bb 100644
--- a/pkg/execution/a_execution-packr.go
+++ b/pkg/execution/a_execution-packr.go
@@ -11,6 +11,7 @@ func init() {
 	_ = packr.PackJSONBytes("./graphql_definitions", "directives/http_json_datasource.graphql", "\"ZGlyZWN0aXZlIEBIdHRwSnNvbkRhdGFTb3VyY2UgKAogICAgIiIiCiAgICB0aGUgaG9zdCBvZiB0aGUgb3JpZ2luCiAgICAiIiIKICAgIGhvc3Q6IFN0cmluZyEKICAgICIiIgogICAgdGhlIHVybCBvZiB0aGUgb3JpZ2luCiAgICAiIiIKICAgIHVybDogU3RyaW5nIQogICAgIiIiCiAgICB0aGUgSFRUUCBtZXRob2QgdGhlIGNsaWVudCBzaG91bGQgdXNlIGZvciB0aGUgcmVxdWVzdCwgZGVmYXVsdHMgdG8gR0VUCiAgICAiIiIKICAgIG1ldGhvZDogSFRUUF9NRVRIT0QgPSBHRVQKICAgICIiIgogICAgcGFyYW1ldGVycwogICAgIiIiCiAgICBwYXJhbXM6IFtQYXJhbWV0ZXJdCiAgICAiIiIKICAgIGJvZHkgaXMgdGhlIG9wdGlvbmFsIGRlZmluaXRpb24gb2YgdGhlIGh0dHAgYm9keSB0byBiZSBzZW50IHRvIHRoZSB1cHN0cmVhbQogICAgZ29sYW5nIHRlbXBsYXRpbmcgc3ludGF4IG1pZ2h0IGJlIHVzZWQgdG8gY3JlYXRlIHRoZSBib2R5CiAgICAiIiIKICAgIGJvZHk6IFN0cmluZwogICAgIiIiCiAgICBoZWFkZXJzIGFyZSB0aGUga2V5IHZhbHVlIHBhaXJzIHRvIGJlIHNldCBvbiB0aGUgdXBzdHJlYW0gcmVxdWVzdAogICAgIiIiCiAgICBoZWFkZXJzOiBbSGVhZGVyXQopIG9uIEZJRUxEX0RFRklOSVRJT04=\"")
 	_ = packr.PackJSONBytes("./graphql_definitions", "directives/http_polling_stream_datasource.graphql", "\"IiIiCkh0dHBQb2xsaW5nU3RyZWFtRGF0YVNvdXJjZQoiIiIKZGlyZWN0aXZlIEBIdHRwUG9sbGluZ1N0cmVhbURhdGFTb3VyY2UgKAogICAgIiIiCiAgICBob3N0IGlzIHRoZSBob3N0IG5hbWUgb2YgdGhlIGRhdGEgc291cmNlLCBlLmcuIGV4YW1wbGUuY29tCiAgICAiIiIKICAgIGhvc3Q6IFN0cmluZyEKICAgICIiIgogICAgdXJsIGlzIHRoZSBVUkwgZnJvbSB3aGljaCB0aGUgZGF0YSBzb3VyY2Ugc2hvdWxkIHBvbGwgZGF0YSBmcm9tCiAgICB5b3UgY291bGQgdXNlIGdvbGFuZyB0ZW1wbGF0aW5nIHN5bnRheCwgZS5nLjoKICAgICcvdXNlcnMve3sgLklkIH19JyArIHBhcmFtZXRlciBuYW1lZCBpZCB3aXRoIHZhbHVlIDEgKHNlZSBiZWxvdykgd2lsbCByZW5kZXIgdG8gVVJMICcvdXNlcnMvMScKICAgICIiIgogICAgdXJsOiBTdHJpbmchCiAgICAibWV0aG9kIGlzIHRoZSBIVFRQIG1ldGhvZCB0aGUgZGF0YSBzb3VyY2Ugc2hvdWxkIHVzZSB0byBmZXRjaCBkYXRhIGZyb20gdGhlIHVwc3RyZWFtIgogICAgbWV0aG9kOiBIVFRQX01FVEhPRCA9IEdFVAogICAgInRoZSBkZWxheSBpbiBzZWNvbmRzIGJldHdlZW4gZWFjaCBwb2xsaW5nIgogICAgZGVsYXlTZWNvbmRzOiBJbnQgPSA1CiAgICAiIiIKICAgIHBhcmFtcyBhcmUgdGhlIHBhcmFtZXRlcnMgdGhhdCBzaG91bGQgZ2V0IHBhc3NlZCB0byB0aGUgZGF0YSBzb3VyY2UKICAgIHlvdSBjb3VsZCB1c2UgdGhpcyBmdW5jdGlvbiB0byBwYXNzIHZhcmlhYmxlcyBmcm9tIGZpZWxkIHZhcmlhYmxlcywgdGhlIGNvbnRleHQgb3IgcGFyZW50IG9iamVjdCB0byB0aGUgZGF0YSBzb3VyY2UKICAgIHRoc2UgY291bGQgYmUgdXNlZCB0byAncmVuZGVyJyB0aGUgdXJsIGR5bmFtaWNhbGx5CiAgICAiIiIKICAgIHBhcmFtczogW1BhcmFtZXRlcl0KKSBvbiBGSUVMRF9ERUZJTklUSU9O\"")
 	_ = packr.PackJSONBytes("./graphql_definitions", "directives/mapping.graphql", "\"IiIiCm1hcHBpbmcgaXMgdGhlIGRpcmVjdGl2ZSB0byBkZWZpbmUgbWFwcGluZ3MgZnJvbSByZXNwb25zZSBvYmplY3RzIHRvIEdyYXBoUUwgZmllbGRzCiIiIgpkaXJlY3RpdmUgQG1hcHBpbmcoCiAgICAiIiIKICAgIG1vZGUgZGVmaW5lcyB0aGUgb3BlcmF0aW5nIG1vZGUgb2YgdGhlIG1hcHBpbmcKICAgICIiIgogICAgbW9kZTogTUFQUElOR19NT0RFISA9IFBBVEhfU0VMRUNUT1IKICAgICIiIgogICAgcGF0aFNlbGVjdG9yIGlzIHRoZSBvcHRpb25hbCBzZWxlY3RvciBTdHJpbmcgdG8gY3VzdG9taXplIHRoZSBtYXBwaW5ncwogICAgc2VlIGh0dHBzOi8vZ2l0aHViLmNvbS90aWR3YWxsL2dqc29uIGZvciBtb3JlIGluZm9ybWF0aW9uIG9uIHdoYXQgc3ludGF4IGlzIHBvc3NpYmxlCiAgICAiIiIKICAgIHBhdGhTZWxlY3RvcjogU3RyaW5nCikgb24gRklFTERfREVGSU5JVElPTg==\"")
+	_ = packr.PackJSONBytes("./graphql_definitions", "directives/nats_datasource.graphql", "\"IiIiCk5hdHNEYXRhU291cmNlCiIiIgpkaXJlY3RpdmUgQE5hdHNEYXRhU291cmNlICgKICAgIGFkZHI6IFN0cmluZyEKICAgIHRvcGljOiBTdHJpbmchCikgb24gRklFTERfREVGSU5JVElPTg==\"")
 	_ = packr.PackJSONBytes("./graphql_definitions", "directives/static_datasource.graphql", "\"ZGlyZWN0aXZlIEBTdGF0aWNEYXRhU291cmNlICgKICAgIGRhdGE6IFN0cmluZwopIG9uIEZJRUxEX0RFRklOSVRJT04=\"")
 	_ = packr.PackJSONBytes("./graphql_definitions", "enums/http_method.graphql", "\"ZW51bSBIVFRQX01FVEhPRCB7CiAgICBHRVQKICAgIFBPU1QKICAgIFVQREFURQogICAgREVMRVRFCn0=\"")
 	_ = packr.PackJSONBytes("./graphql_definitions", "enums/mapping_mode.graphql", "\"IiIiCk1BUFBJTkdfTU9ERSBpcyB0aGUgZW51bSB0byBkZWZpbmUgdGhlIGRpZmZlcmVudCBtYXBwZXIgbW9kZXMKIiIiCmVudW0gTUFQUElOR19NT0RFIHsKICAgICIiIgogICAgTk9ORSBpcyB0aGUgbW9kZSB0byBza2lwIG1hcHBpbmdzIGF0IGFsbCwgdGhlcmVmb3JlIHRoZSBwYXJlbnQgb2JqZWN0IHdpbGwgZGlyZWN0bHkgZ2V0IHBhc3NlZCB0byB0aGUgY2hpbGQKICAgICIiIgogICAgTk9ORQogICAgIiIiCiAgICBQQVRIX1NFTEVDVE9SIGlzIHRoZSBtb2RlIHdoaWNoIGFwcGxpZXMgYSBwYXRoIHNlbGVjdG9yIHN0cmluZyBvbiB0aGUgcGFyZW50IG9iamVjdCBiZWZvcmUgcGFzc2luZyBpdCB0byB0aGUgY2hpbGQKICAgIHNlZSBodHRwczovL2dpdGh1Yi5jb20vdGlkd2FsbC9nanNvbiBmb3IgbW9yZSBpbmZvcm1hdGlvbiBvbiB3aGF0IHN5bnRheCBpcyBwb3NzaWJsZQogICAgIiIiCiAgICBQQVRIX1NFTEVDVE9SCn0=\"")
diff --git a/pkg/execution/datasource.go b/pkg/execution/datasource.go
index ca324052..a1492fc5 100644
--- a/pkg/execution/datasource.go
+++ b/pkg/execution/datasource.go
@@ -75,4 +75,22 @@ type BaseDataSourcePlanner struct {
 	definition, operation *ast.Document      // nolint
 	args                  []Argument         // nolint
 	graphqlDefinitions    *packr.Box         // nolint
+	rootField             rootField          // nolint
 }
+
+type rootField struct {
+	isDefined bool
+	ref       int
+}
+
+func (r *rootField) setIfNotDefined(ref int){
+	if r.isDefined {
+		return
+	}
+	r.isDefined = true
+	r.ref = ref
+}
+
+func (r *rootField) isDefinedAndEquals(ref int) bool {
+	return r.isDefined && r.ref == ref
+}
\ No newline at end of file
diff --git a/pkg/execution/datasource_nats.go b/pkg/execution/datasource_nats.go
new file mode 100644
index 00000000..6abde6f1
--- /dev/null
+++ b/pkg/execution/datasource_nats.go
@@ -0,0 +1,178 @@
+package execution
+
+import (
+	"github.com/jensneuse/graphql-go-tools/pkg/ast"
+	"github.com/jensneuse/graphql-go-tools/pkg/astvisitor"
+	"github.com/jensneuse/graphql-go-tools/pkg/lexer/literal"
+	"github.com/nats-io/nats.go"
+	"go.uber.org/zap"
+	"io"
+	"sync"
+	"time"
+)
+
+type NatsDataSourcePlanner struct {
+	BaseDataSourcePlanner
+}
+
+func NewNatsDataSourcePlanner(baseDataSourcePlanner BaseDataSourcePlanner) *NatsDataSourcePlanner {
+	return &NatsDataSourcePlanner{
+		BaseDataSourcePlanner: baseDataSourcePlanner,
+	}
+}
+
+func (n *NatsDataSourcePlanner) DirectiveName() []byte {
+	return []byte("NatsDataSource")
+}
+
+func (n *NatsDataSourcePlanner) DirectiveDefinition() []byte {
+	data, _ := n.graphqlDefinitions.Find("directives/nats_datasource.graphql")
+	return data
+}
+
+func (n *NatsDataSourcePlanner) Plan() (DataSource, []Argument) {
+	return &NatsDataSource{
+		log: n.log,
+	}, n.args
+}
+
+func (n *NatsDataSourcePlanner) Initialize(walker *astvisitor.Walker, operation, definition *ast.Document, args []Argument, resolverParameters []ResolverParameter) {
+	n.walker, n.operation, n.definition, n.args = walker, operation, definition, args
+}
+
+func (n *NatsDataSourcePlanner) EnterInlineFragment(ref int) {
+
+}
+
+func (n *NatsDataSourcePlanner) LeaveInlineFragment(ref int) {
+
+}
+
+func (n *NatsDataSourcePlanner) EnterSelectionSet(ref int) {
+
+}
+
+func (n *NatsDataSourcePlanner) LeaveSelectionSet(ref int) {
+
+}
+
+func (n *NatsDataSourcePlanner) EnterField(ref int) {
+	n.rootField.setIfNotDefined(ref)
+}
+
+func (n *NatsDataSourcePlanner) LeaveField(ref int) {
+	if !n.rootField.isDefinedAndEquals(ref) {
+		return
+	}
+	definition, exists := n.walker.FieldDefinition(ref)
+	if !exists {
+		return
+	}
+	directive, exists := n.definition.FieldDefinitionDirectiveByName(definition, n.DirectiveName())
+	if !exists {
+		return
+	}
+	value, exists := n.definition.DirectiveArgumentValueByName(directive, literal.ADDR)
+	if !exists {
+		return
+	}
+	variableValue := n.definition.StringValueContentBytes(value.Ref)
+	arg := &StaticVariableArgument{
+		Name:  literal.ADDR,
+		Value: make([]byte, len(variableValue)),
+	}
+	copy(arg.Value, variableValue)
+	n.args = append(n.args, arg)
+
+	value, exists = n.definition.DirectiveArgumentValueByName(directive, literal.TOPIC)
+	if !exists {
+		return
+	}
+	variableValue = n.definition.StringValueContentBytes(value.Ref)
+	arg = &StaticVariableArgument{
+		Name:  literal.TOPIC,
+		Value: make([]byte, len(variableValue)),
+	}
+	copy(arg.Value, variableValue)
+	n.args = append(n.args, arg)
+}
+
+type NatsDataSource struct {
+	log  *zap.Logger
+	conn *nats.Conn
+	sub  *nats.Subscription
+	once sync.Once
+}
+
+func (n *NatsDataSource) Resolve(ctx Context, args ResolvedArgs, out io.Writer) Instruction {
+	var err error
+	n.once.Do(func() {
+
+		addrArg := args.ByKey(literal.ADDR)
+		topicArg := args.ByKey(literal.TOPIC)
+
+		addr := nats.DefaultURL
+		topic := string(topicArg)
+
+		if len(addrArg) != 0 {
+			addr = string(addrArg)
+		}
+
+		go func() {
+			<-ctx.Done()
+			if n.sub != nil {
+				n.log.Debug("NatsDataSource.unsubscribing",
+					zap.String("addr", addr),
+					zap.String("topic", topic),
+				)
+				err := n.sub.Unsubscribe()
+				if err != nil {
+					n.log.Error("Unsubscribe", zap.Error(err))
+				}
+			}
+			if n.conn != nil {
+				n.log.Debug("NatsDataSource.closing",
+					zap.String("addr", addr),
+					zap.String("topic", topic),
+				)
+				n.conn.Close()
+			}
+		}()
+
+		n.log.Debug("NatsDataSource.connecting",
+			zap.String("addr", addr),
+			zap.String("topic", topic),
+		)
+
+		n.conn, err = nats.Connect(addr)
+		if err != nil {
+			panic(err)
+		}
+
+		n.log.Debug("NatsDataSource.subscribing",
+			zap.String("addr", addr),
+			zap.String("topic", topic),
+		)
+
+		n.sub, err = n.conn.SubscribeSync(topic)
+		if err != nil {
+			panic(err)
+		}
+	})
+
+	if err != nil {
+		return CloseConnection
+	}
+
+	message, err := n.sub.NextMsg(time.Minute)
+	if err != nil {
+		return CloseConnection
+	}
+
+	_, err = out.Write(message.Data)
+	if err != nil {
+		return CloseConnection
+	}
+
+	return KeepStreamAlive
+}
diff --git a/pkg/execution/execution.go b/pkg/execution/execution.go
index 079d425e..42aa6daf 100644
--- a/pkg/execution/execution.go
+++ b/pkg/execution/execution.go
@@ -158,15 +158,14 @@ func (e *Executor) resolveNode(node Node, data []byte, path string, prefetch *sy
 			data = unsafebytes.StringToBytes(result.Raw)
 		}
 
-		if e.err == jsonparser.KeyPathNotFoundError {
-			e.err = nil
+		if len(data) == 0 || bytes.Equal(data, literal.NULL) {
 			e.write(literal.NULL)
 			return
 		}
 		if node.QuoteValue {
 			e.write(literal.QUOTE)
 		}
-		e.write(data)
+		e.write(e.escape(data))
 		if node.QuoteValue {
 			e.write(literal.QUOTE)
 		}
@@ -183,10 +182,6 @@ func (e *Executor) resolveNode(node Node, data []byte, path string, prefetch *sy
 			}
 		}
 
-		/*_, e.err = jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, err error) {
-			listItems = append(listItems, value)
-		}, node.Path...)*/
-
 		var result []gjson.Result
 		if node.PathSelector.Path == "" {
 			result = gjson.ParseBytes(data).Array()
@@ -240,6 +235,28 @@ func (e *Executor) resolveNode(node Node, data []byte, path string, prefetch *sy
 	}
 }
 
+func (e *Executor) escape(data []byte) []byte {
+	if !e.dataNeedsEscaping(data) {
+		return data
+	}
+	out, err := json.Marshal(unsafebytes.BytesToString(data))
+	if err != nil {
+		panic(err)
+	}
+	out = out[1 : len(out)-1]
+	return out
+}
+
+func (e *Executor) dataNeedsEscaping(data []byte) bool {
+	for i := range data {
+		switch data[i] {
+		case runes.BACKSLASH, runes.TAB, runes.SPACE, runes.LINETERMINATOR:
+			return true
+		}
+	}
+	return false
+}
+
 func (e *Executor) ResolveArgs(args []Argument, data []byte) ResolvedArgs {
 
 	args = append(args, e.context.ExtraArguments...)
diff --git a/pkg/execution/fixtures/introspection_execution.golden b/pkg/execution/fixtures/introspection_execution.golden
new file mode 100644
index 00000000..1160a4a5
--- /dev/null
+++ b/pkg/execution/fixtures/introspection_execution.golden
@@ -0,0 +1,1 @@
+{"data":{"__schema":{"queryType":{"name":"Query"},"mutationType":null,"subscriptionType":null,"types":[{"kind":"OBJECT","name":"Query","description":null,"fields":[{"name":"foo","description":"multiline\n\t\t\tdescription","args":[],"type":{"kind":"SCALAR","name":"String","ofType":null},"isDeprecated":null,"deprecationReason":null}],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]},{"kind":"SCALAR","name":"Int","description":null,"fields":[],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]},{"kind":"SCALAR","name":"Float","description":null,"fields":[],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]},{"kind":"SCALAR","name":"String","description":null,"fields":[],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]},{"kind":"SCALAR","name":"Boolean","description":null,"fields":[],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]},{"kind":"SCALAR","name":"ID","description":null,"fields":[],"inputFields":[],"interfaces":[],"enumValues":[],"possibleTypes":[]}],"directives":[{"name":"include","description":"Directs the executor to include this field or fragment only when the argument is true.","locations":["FIELD","FRAGMENT_SPREAD","INLINE_FRAGMENT"],"args":[{"name":"if","description":"Included when true.","type":{"kind":"NON_NULL","name":null,"ofType":{"kind":"SCALAR","name":"Boolean","ofType":null}},"defaultValue":null}]},{"name":"skip","description":"Directs the executor to skip this field or fragment when the argument is true.","locations":["FIELD","FRAGMENT_SPREAD","INLINE_FRAGMENT"],"args":[{"name":"if","description":"Skipped when true.","type":{"kind":"NON_NULL","name":null,"ofType":{"kind":"SCALAR","name":"Boolean","ofType":null}},"defaultValue":null}]},{"name":"deprecated","description":"Marks an element of a GraphQL schema as no longer supported.","locations":["FIELD_DEFINITION","ENUM_VALUE"],"args":[{"name":"reason","description":"Explains why this element was deprecated, usually also including a suggestion\n    for how to access supported similar data. Formatted in\n    [Markdown](https://daringfireball.net/projects/markdown/).","type":{"kind":"SCALAR","name":"String","ofType":null},"defaultValue":"\"No longer supported\""}]}]}}}
\ No newline at end of file
diff --git a/pkg/execution/fixtures/render_graphql_definitions.golden b/pkg/execution/fixtures/render_graphql_definitions.golden
index f805baba..2570c856 100644
--- a/pkg/execution/fixtures/render_graphql_definitions.golden
+++ b/pkg/execution/fixtures/render_graphql_definitions.golden
@@ -86,6 +86,14 @@ directive @mapping
     pathSelector: String
  on FIELD_DEFINITION
 
+"""
+NatsDataSource
+"""
+directive @NatsDataSource
+    addr: String! 
+    topic: String!
+ on FIELD_DEFINITION
+
 directive @StaticDataSource
     data: String
  on FIELD_DEFINITION
diff --git a/pkg/execution/graphql_definitions/directives/nats_datasource.graphql b/pkg/execution/graphql_definitions/directives/nats_datasource.graphql
new file mode 100644
index 00000000..1fb815ec
--- /dev/null
+++ b/pkg/execution/graphql_definitions/directives/nats_datasource.graphql
@@ -0,0 +1,7 @@
+"""
+NatsDataSource
+"""
+directive @NatsDataSource (
+    addr: String!
+    topic: String!
+) on FIELD_DEFINITION
\ No newline at end of file
diff --git a/pkg/execution/handler.go b/pkg/execution/handler.go
index 02983083..4c6f2cd8 100644
--- a/pkg/execution/handler.go
+++ b/pkg/execution/handler.go
@@ -123,7 +123,7 @@ func (h *Handler) Handle(requestData, extraVariables []byte) (executor *Executor
 
 	executor = NewExecutor()
 	ctx = Context{
-		Variables: variables,
+		Variables:      variables,
 		ExtraArguments: extraArguments,
 	}
 
@@ -185,6 +185,9 @@ func (h *Handler) resolverDefinitions(report *operationreport.Report) ResolverDe
 			func() DataSourcePlanner {
 				return NewTypeDataSourcePlanner(baseDataSourcePlanner)
 			},
+			func() DataSourcePlanner {
+				return NewNatsDataSourcePlanner(baseDataSourcePlanner)
+			},
 		},
 	}
 	walker.RegisterEnterFieldDefinitionVisitor(&visitor)
diff --git a/pkg/lexer/literal/literal.go b/pkg/lexer/literal/literal.go
index 4f866346..52009bb5 100644
--- a/pkg/lexer/literal/literal.go
+++ b/pkg/lexer/literal/literal.go
@@ -21,14 +21,14 @@ var (
 	SUB            = []byte("-")
 	AND            = []byte("&")
 
-	LPAREN = []byte("(")
-	RPAREN = []byte(")")
-	LBRACK = []byte("[")
-	RBRACK = []byte("]")
-	LBRACE = []byte("{")
+	LPAREN        = []byte("(")
+	RPAREN        = []byte(")")
+	LBRACK        = []byte("[")
+	RBRACK        = []byte("]")
+	LBRACE        = []byte("{")
 	DOUBLE_LBRACE = []byte("{{")
 	DOUBLE_RBRACE = []byte("}}")
-	RBRACE = []byte("}")
+	RBRACE        = []byte("}")
 
 	GOBOOL    = []byte("bool")
 	GOINT32   = []byte("int32")
@@ -67,6 +67,8 @@ var (
 	NULL               = []byte("null")
 	DATA               = []byte("data")
 	URL                = []byte("url")
+	ADDR               = []byte("addr")
+	TOPIC              = []byte("topic")
 	HOST               = []byte("host")
 	PARAMS             = []byte("params")
 	FIELD              = []byte("field")


Test patch:
diff --git a/pkg/execution/execution_test.go b/pkg/execution/execution_test.go
index 98d4dfd0..876d6883 100644
--- a/pkg/execution/execution_test.go
+++ b/pkg/execution/execution_test.go
@@ -8,9 +8,10 @@ import (
 	"github.com/cespare/xxhash"
 	"github.com/jensneuse/diffview"
 	"github.com/jensneuse/graphql-go-tools/pkg/ast"
+	"github.com/jensneuse/graphql-go-tools/pkg/introspection"
 	"github.com/jensneuse/graphql-go-tools/pkg/lexer/literal"
+	"github.com/jensneuse/graphql-go-tools/pkg/operationreport"
 	"github.com/sebdah/goldie"
-	"github.com/tidwall/gjson"
 	"go.uber.org/zap"
 	"io"
 	"io/ioutil"
@@ -1238,7 +1239,7 @@ func TestExecutor_GraphqlDataSourceWithParams(t *testing.T) {
 
 	graphQL1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 
-		body,err := httputil.DumpRequest(r,true)
+		body, err := httputil.DumpRequest(r, true)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -1284,7 +1285,7 @@ func TestExecutor_GraphqlDataSourceWithParams(t *testing.T) {
 								},
 							},
 							DataSource: &GraphQLDataSource{
-								log:zap.NewNop(),
+								log: zap.NewNop(),
 							},
 						},
 						BufferName: "countries",
@@ -2043,7 +2044,7 @@ func TestExecutor_HTTPJSONDataSourceWithPathSelector(t *testing.T) {
 									{
 										Name: []byte("firstNames"),
 										Value: &List{
-											PathSelector:PathSelector{
+											PathSelector: PathSelector{
 												Path: "friends.#.first",
 											},
 											Value: &Value{
@@ -2106,22 +2107,57 @@ func prettyJSON(r io.Reader) string {
 	return string(out)
 }
 
-func TestSelectList(t *testing.T){
-	data := []byte(`
-[
-   {
-      "id":2,
-      "name":"Yaara",
-      "birthday":"1990 I guess? ;-)"
-   },
-   {
-      "id":3,
-      "name":"Ahmet",
-      "birthday":"1980"
-   }
-]
-`)
+func TestExecutor_Introspection(t *testing.T) {
+	executor := NewExecutor()
+	ctx := Context{
+		Context: context.Background(),
+	}
+
+	schema := []byte(`
+		schema {
+			query: Query
+		}
+		type Query {
+			"""
+			multiline
+			description
+			"""
+			foo: String
+			__schema: __Schema!
+		}
+	`)
+
+	handler, err := NewHandler(schema, zap.NewNop())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	gen := introspection.NewGenerator()
+	report := operationreport.Report{}
+	data := introspection.Data{}
+	gen.Generate(&handler.definition, &report, &data)
+
+	introspectionData, err := json.Marshal(data)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	out := bytes.Buffer{}
+	_, err = executor.Execute(ctx, introspectionQuery(introspectionData), &out)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	response := out.Bytes()
+
+	goldie.Assert(t,"introspection_execution",response)
+	if t.Failed() {
+
+		fixture, err := ioutil.ReadFile("./fixtures/introspection_execution.golden")
+		if err != nil {
+			t.Fatal(err)
+		}
 
-	result := gjson.ParseBytes(data)
-	_ = result
-}
\ No newline at end of file
+		diffview.NewGoland().DiffViewBytes("execution", fixture, response)
+	}
+}
diff --git a/pkg/execution/planning_test.go b/pkg/execution/planning_test.go
index 1697b589..c47e90a6 100644
--- a/pkg/execution/planning_test.go
+++ b/pkg/execution/planning_test.go
@@ -3336,6 +3336,448 @@ enum __TypeKind {
 `
 }
 
+func introspectionQuery(schema []byte) RootNode {
+	return &Object{
+		operationType: ast.OperationTypeQuery,
+		Fields: []Field{
+			{
+				Name: []byte("data"),
+				Value: &Object{
+					Fetch: &SingleFetch{
+						Source: &DataSourceInvocation{
+							DataSource: &SchemaDataSource{
+								schemaBytes: schema,
+							},
+						},
+						BufferName: "__schema",
+					},
+					Fields: []Field{
+						{
+							Name:        []byte("__schema"),
+							HasResolver: true,
+							Value: &Object{
+								PathSelector: PathSelector{
+									Path: "__schema",
+								},
+								Fields: []Field{
+									{
+										Name: []byte("queryType"),
+										Value: &Object{
+											PathSelector: PathSelector{
+												Path: "queryType",
+											},
+											Fields: []Field{
+												{
+													Name: []byte("name"),
+													Value: &Value{
+														PathSelector: PathSelector{
+															Path: "name",
+														},
+														QuoteValue: true,
+													},
+												},
+											},
+										},
+									},
+									{
+										Name: []byte("mutationType"),
+										Value: &Object{
+											PathSelector: PathSelector{
+												Path: "mutationType",
+											},
+											Fields: []Field{
+												{
+													Name: []byte("name"),
+													Value: &Value{
+														PathSelector: PathSelector{
+															Path: "name",
+														},
+														QuoteValue: true,
+													},
+												},
+											},
+										},
+									},
+									{
+										Name: []byte("subscriptionType"),
+										Value: &Object{
+											PathSelector: PathSelector{
+												Path: "subscriptionType",
+											},
+											Fields: []Field{
+												{
+													Name: []byte("name"),
+													Value: &Value{
+														PathSelector: PathSelector{
+															Path: "name",
+														},
+														QuoteValue: true,
+													},
+												},
+											},
+										},
+									},
+									{
+										Name: []byte("types"),
+										Value: &List{
+											PathSelector: PathSelector{
+												Path: "types",
+											},
+											Value: &Object{
+												Fields: []Field{
+													{
+														Name: []byte("kind"),
+														Value: &Value{
+															PathSelector: PathSelector{
+																Path: "kind",
+															},
+															QuoteValue: true,
+														},
+													},
+													{
+														Name: []byte("name"),
+														Value: &Value{
+															PathSelector: PathSelector{
+																Path: "name",
+															},
+															QuoteValue: true,
+														},
+													},
+													{
+														Name: []byte("description"),
+														Value: &Value{
+															PathSelector: PathSelector{
+																Path: "description",
+															},
+															QuoteValue: true,
+														},
+													},
+													{
+														Name: []byte("fields"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "fields",
+															},
+															Value: &Object{
+																Fields: []Field{
+																	{
+																		Name: []byte("name"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "name",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("description"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "description",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("args"),
+																		Value: &List{
+																			PathSelector: PathSelector{
+																				Path: "args",
+																			},
+																			Value: &Object{
+																				Fields: []Field{
+																					{
+																						Name: []byte("name"),
+																						Value: &Value{
+																							PathSelector: PathSelector{
+																								Path: "name",
+																							},
+																							QuoteValue: true,
+																						},
+																					},
+																					{
+																						Name: []byte("description"),
+																						Value: &Value{
+																							PathSelector: PathSelector{
+																								Path: "description",
+																							},
+																							QuoteValue: true,
+																						},
+																					},
+																					{
+																						Name: []byte("type"),
+																						Value: &Object{
+																							PathSelector: PathSelector{
+																								Path: "type",
+																							},
+																							Fields: kindNameDeepFields,
+																						},
+																					},
+																					{
+																						Name: []byte("defaultValue"),
+																						Value: &Value{
+																							PathSelector: PathSelector{
+																								Path: "defaultValue",
+																							},
+																							QuoteValue: true,
+																						},
+																					},
+																				},
+																			},
+																		},
+																	},
+																	{
+																		Name: []byte("type"),
+																		Value: &Object{
+																			PathSelector: PathSelector{
+																				Path: "type",
+																			},
+																			Fields: kindNameDeepFields,
+																		},
+																	},
+																	{
+																		Name: []byte("isDeprecated"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "isDeprecated",
+																			},
+																			QuoteValue: false,
+																		},
+																	},
+																	{
+																		Name: []byte("deprecationReason"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "deprecationReason",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																},
+															},
+														},
+													},
+													{
+														Name: []byte("inputFields"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "inputFields",
+															},
+															Value: &Object{
+																Fields: []Field{
+																	{
+																		Name: []byte("name"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "name",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("description"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "description",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("type"),
+																		Value: &Object{
+																			PathSelector: PathSelector{
+																				Path: "type",
+																			},
+																			Fields: kindNameDeepFields,
+																		},
+																	},
+																	{
+																		Name: []byte("defaultValue"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "defaultValue",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																},
+															},
+														},
+													},
+													{
+														Name: []byte("interfaces"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "interfaces",
+															},
+															Value: &Object{
+																Fields: kindNameDeepFields,
+															},
+														},
+													},
+													{
+														Name: []byte("enumValues"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "enumValues",
+															},
+															Value: &Object{
+																Fields: []Field{
+																	{
+																		Name: []byte("name"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "name",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("description"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "description",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("isDeprecated"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "isDeprecated",
+																			},
+																			QuoteValue: false,
+																		},
+																	},
+																	{
+																		Name: []byte("deprecationReason"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "deprecationReason",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																},
+															},
+														},
+													},
+													{
+														Name: []byte("possibleTypes"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "possibleTypes",
+															},
+															Value: &Object{
+																Fields: kindNameDeepFields,
+															},
+														},
+													},
+												},
+											},
+										},
+									},
+									{
+										Name: []byte("directives"),
+										Value: &List{
+											PathSelector: PathSelector{
+												Path: "directives",
+											},
+											Value: &Object{
+												Fields: []Field{
+													{
+														Name: []byte("name"),
+														Value: &Value{
+															PathSelector: PathSelector{
+																Path: "name",
+															},
+															QuoteValue: true,
+														},
+													},
+													{
+														Name: []byte("description"),
+														Value: &Value{
+															PathSelector: PathSelector{
+																Path: "description",
+															},
+															QuoteValue: true,
+														},
+													},
+													{
+														Name: []byte("locations"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "locations",
+															},
+															Value: &Value{
+																QuoteValue: true,
+															},
+														},
+													},
+													{
+														Name: []byte("args"),
+														Value: &List{
+															PathSelector: PathSelector{
+																Path: "args",
+															},
+															Value: &Object{
+																Fields: []Field{
+																	{
+																		Name: []byte("name"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "name",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("description"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "description",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																	{
+																		Name: []byte("type"),
+																		Value: &Object{
+																			PathSelector: PathSelector{
+																				Path: "type",
+																			},
+																			Fields: kindNameDeepFields,
+																		},
+																	},
+																	{
+																		Name: []byte("defaultValue"),
+																		Value: &Value{
+																			PathSelector: PathSelector{
+																				Path: "defaultValue",
+																			},
+																			QuoteValue: true,
+																		},
+																	},
+																},
+															},
+														},
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 var kindNameDeepFields = []Field{
 	{
 		Name: []byte("kind"),