feat: edfs refactor kafka and nats as datasource in router

demo/cmd/mood/main.go

@@ -3,11 +3,12 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood"
 	"log"
 	"net/http"
 	"os"
 
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood"
+
 	"github.com/99designs/gqlgen/graphql"
 	"github.com/99designs/gqlgen/graphql/handler/debug"
 	"github.com/99designs/gqlgen/graphql/playground"
@@ -31,7 +32,9 @@ func main() {
 		port = defaultPort
 	}
 
-	srv := subgraphs.NewDemoServer(mood.NewSchema(nil))
+	srv := subgraphs.NewDemoServer(mood.NewSchema(nil, func(name string) string {
+		return name
+	}))
 
 	srv.Use(&debug.Tracer{})
 	srv.Use(otelgqlgen.Middleware(otelgqlgen.WithCreateSpanFromFields(func(ctx *graphql.FieldContext) bool {

demo/pkg/subgraphs/availability/availability.go

@@ -2,13 +2,13 @@ package availability
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/availability/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/availability/subgraph/generated"
 )
 
-func NewSchema(pubSubBySourceName map[string]pubsub_datasource.NatsPubSub, pubSubName func(string) string) graphql.ExecutableSchema {
+func NewSchema(pubSubBySourceName map[string]nats.AdapterInterface, pubSubName func(string) string) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: pubSubBySourceName,
 		GetPubSubName:          pubSubName,

demo/pkg/subgraphs/availability/subgraph/resolver.go

@@ -1,14 +1,14 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 	GetPubSubName          func(string) string
 }

demo/pkg/subgraphs/countries/countries.go

@@ -2,13 +2,13 @@ package countries
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/countries/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/countries/subgraph/generated"
 )
 
-func NewSchema(pubSubBySourceName map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(pubSubBySourceName map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: pubSubBySourceName,
 	}})

demo/pkg/subgraphs/countries/subgraph/resolver.go

@@ -1,8 +1,9 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -11,5 +12,5 @@ import (
 
 type Resolver struct {
 	mux                    sync.Mutex
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 }

demo/pkg/subgraphs/employees/employees.go

@@ -2,13 +2,12 @@ package employees
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
-
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph/generated"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 		EmployeesData:          subgraph.Employees,

demo/pkg/subgraphs/employees/subgraph/resolver.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -15,7 +15,7 @@ import (
 
 type Resolver struct {
 	mux                    sync.Mutex
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 	EmployeesData          []*model.Employee
 }
 

demo/pkg/subgraphs/family/family.go

@@ -2,13 +2,13 @@ package family
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/family/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/family/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 	}})

demo/pkg/subgraphs/family/subgraph/resolver.go

@@ -1,13 +1,13 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 }

demo/pkg/subgraphs/hobbies/hobbies.go

@@ -2,13 +2,13 @@ package hobbies
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 	}})

demo/pkg/subgraphs/hobbies/subgraph/resolver.go

@@ -1,18 +1,18 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"reflect"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 }
 
 func (r *Resolver) Employees(hobby model.Hobby) ([]*model.Employee, error) {

demo/pkg/subgraphs/mood/mood.go

@@ -2,14 +2,15 @@ package mood
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface, getPubSubName func(string) string) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
+		GetPubSubName:          getPubSubName,
 	}})
 }

demo/pkg/subgraphs/mood/subgraph/resolver.go

@@ -1,14 +1,14 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.AdapterInterface
 	GetPubSubName          func(string) string
 }

demo/pkg/subgraphs/products/products.go

@@ -2,13 +2,13 @@ package products
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID:       natsPubSubByProviderID,
 		TopSecretFederationFactsData: subgraph.TopSecretFederationFacts,

demo/pkg/subgraphs/products/subgraph/resolver.go

@@ -1,9 +1,10 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -12,6 +13,6 @@ import (
 
 type Resolver struct {
 	mux                          sync.Mutex
-	NatsPubSubByProviderID       map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID       map[string]nats.AdapterInterface
 	TopSecretFederationFactsData []model.TopSecretFact
 }

demo/pkg/subgraphs/products_fg/products.go

@@ -2,13 +2,13 @@ package products_fg
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.AdapterInterface) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID:       natsPubSubByProviderID,
 		TopSecretFederationFactsData: subgraph.TopSecretFederationFacts,

demo/pkg/subgraphs/products_fg/subgraph/resolver.go

@@ -1,9 +1,10 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -12,6 +13,6 @@ import (
 
 type Resolver struct {
 	mux                          sync.Mutex
-	NatsPubSubByProviderID       map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID       map[string]nats.AdapterInterface
 	TopSecretFederationFactsData []model.TopSecretFact
 }