Merge pull request #8 from Zondax/ezequiel/new_connections

Wrappers for connection types

Author: jleni

Makefile

@@ -1,11 +1,5 @@
-
-# Constants to override at build time
-PACKAGE := github.com/zondax/zindexer
-REVISION := $(shell git rev-parse --short HEAD)
-APPNAME := zindexer
-
 build:
-	go build -ldflags "-X $(PACKAGE).GitRevision=$(REVISION)" -o $(APPNAME) ./...
+	go build ./...
 
 clean:
 	go clean

cli.go

@@ -16,7 +16,7 @@ type DefaultConfigHandler func()
 var defaultConfigHandler DefaultConfigHandler
 
 func SetupCloseHandler(handler CleanUpHandler) {
-	c := make(chan os.Signal)
+	c := make(chan os.Signal, 1)
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
 	go func() {
 		<-c

config.go

@@ -13,21 +13,17 @@ type DBConnectionParams struct {
 	Port     string
 }
 
+type GraphqlClientParams struct {
+	Host  string
+	Token string
+}
+
 type DBConnectionConfig struct {
-	gorm *gorm.Config
+	Gorm *gorm.Config
 }
 
 func (p *DBConnectionParams) GetDSN() (string, error) {
 	return fmt.Sprintf(
 		"user=%s password=%s dbname=%s host=%s port=%s sslmode=disable",
 		p.User, p.Password, p.Name, p.Host, p.Port), nil
 }
-
-func ConnectDB(params DBConnectionParams, config DBConnectionConfig) (*gorm.DB, error) {
-	conn, err := NewPostgresConnection(&params, &config)
-	if err != nil {
-		return nil, err
-	}
-
-	return conn.GetDB(), nil
-}

connection.go

@@ -0,0 +1,109 @@
+package connections
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/hasura/go-graphql-client"
+	"go.uber.org/zap"
+	"golang.org/x/oauth2"
+	"time"
+)
+
+const ConnectTimeout = 10 * time.Second
+
+type GraphqlClient struct {
+	Host   string
+	Client *graphql.Client
+}
+
+type GraphqlSubscriptionClient struct {
+	Client *graphql.SubscriptionClient
+	Id     string
+}
+
+func NewGraphqlQueryClient(host string, token string) GraphqlClient {
+	src := oauth2.StaticTokenSource(
+		&oauth2.Token{AccessToken: token},
+	)
+	httpClient := oauth2.NewClient(context.Background(), src)
+	return GraphqlClient{
+		Host:   host,
+		Client: graphql.NewClient(host, httpClient),
+	}
+}
+
+func (c GraphqlClient) Connect() error {
+	return nil
+}
+
+func NewGraphqlSubscriptionClient(host string, token string) (error, GraphqlSubscriptionClient) {
+	client := graphql.NewSubscriptionClient(host).
+		WithConnectionParams(map[string]interface{}{
+			"headers": map[string]string{
+				"x-hasura-admin-secret": token,
+			},
+		}).OnError(onClientError)
+
+	return nil, GraphqlSubscriptionClient{Client: client}
+}
+
+func onClientError(sc *graphql.SubscriptionClient, err error) error {
+	zap.S().Fatalf("Connection error on subscription client: %s", err.Error())
+	return err
+}
+
+func (c GraphqlSubscriptionClient) Subscribe(query interface{}, handler func(message *json.RawMessage, err error) error) error {
+	id, err := c.Client.Subscribe(query, nil, handler)
+	if err != nil {
+		return err
+	}
+	c.Id = id
+	return nil
+}
+
+func (c GraphqlSubscriptionClient) Unsubscribe() error {
+	err := c.Client.Unsubscribe(c.Id)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c GraphqlSubscriptionClient) Start() error {
+	errCh := make(chan error)
+	readyCh := make(chan bool)
+
+	c.Client.OnConnected(func() {
+		readyCh <- true
+	})
+
+	go func() {
+		err := c.Client.Run()
+		if err != nil {
+			errCh <- err
+		}
+	}()
+
+	for {
+		select {
+		case err := <-errCh:
+			close(errCh)
+			close(readyCh)
+			return err
+		case <-readyCh:
+			close(readyCh)
+			return nil
+		case <-time.After(ConnectTimeout):
+			return fmt.Errorf("timeout while waiting subscriber client to connect to host")
+		}
+	}
+}
+
+func (c GraphqlSubscriptionClient) Stop() error {
+	err := c.Client.Close()
+	if err != nil {
+		return err
+	}
+	return nil
+}

connections/graphql.go

@@ -0,0 +1,38 @@
+package connections
+
+import (
+	"fmt"
+	"github.com/Zondax/zindexer"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func NewPostgresConnection(params *zindexer.DBConnectionParams, config *zindexer.DBConnectionConfig) (*GormConnection, error) {
+	dsn, err := params.GetDSN()
+	if err != nil {
+		return nil, fmt.Errorf("failed to retrieve dsn")
+	}
+
+	conn, err := gorm.Open(postgres.Open(dsn), config.Gorm)
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial connect to db '%s@%s:%s': %v", params.Name, params.Host, params.Port, err)
+	}
+
+	return &GormConnection{
+		db: conn,
+	}, nil
+}
+
+func (c *GormConnection) GetDB() *gorm.DB {
+	return c.db
+}
+
+func ConnectDB(params zindexer.DBConnectionParams, config zindexer.DBConnectionConfig) (*gorm.DB, error) {
+	conn, err := NewPostgresConnection(&params, &config)
+	if err != nil {
+		return nil, err
+	}
+
+	return conn.GetDB(), nil
+}

connections/postgres.go

@@ -0,0 +1,58 @@
+package connections
+
+import (
+	"context"
+	"fmt"
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
+)
+
+var ChannelPubSubDefaultConfig = gochannel.Config{
+	OutputChannelBuffer:            10,
+	Persistent:                     true,
+	BlockPublishUntilSubscriberAck: false,
+}
+
+type ChannelPubSub struct {
+	Client *gochannel.GoChannel
+}
+
+// NewPubSubHandlerChannel this handler uses Watermill's go channels
+// implementation as message engine
+func NewPubSubHandlerChannel(config gochannel.Config) ChannelPubSub {
+	if config == (gochannel.Config{}) {
+		// Use default config
+		config = ChannelPubSubDefaultConfig
+	}
+
+	ps := gochannel.NewGoChannel(
+		config,
+		watermill.NewStdLogger(false, false),
+	)
+
+	return ChannelPubSub{
+		Client: ps,
+	}
+}
+
+func (p ChannelPubSub) Subscribe(topic string, cb func(messages <-chan *message.Message)) error {
+	messages, err := p.Client.Subscribe(context.Background(), topic)
+	if err != nil {
+		fmt.Printf("Could not subscribe to topic %s", topic)
+		return err
+	}
+
+	go cb(messages)
+	return nil
+}
+
+func (p ChannelPubSub) Publish(topic string, msg *message.Message) error {
+	err := p.Client.Publish(topic, msg)
+	if err != nil {
+		fmt.Printf("Could not publish to topic %s", topic)
+		return err
+	}
+
+	return nil
+}

connections/pubsub_goChannel.go

@@ -0,0 +1,22 @@
+package connections
+
+import (
+	"github.com/Zondax/zindexer"
+	"gorm.io/gorm"
+)
+
+type GormConnection struct {
+	db *gorm.DB
+}
+
+type DBQueryClient struct {
+	Client zindexer.IDBQueryClient
+}
+
+type DBSubscriptionClient struct {
+	Client zindexer.IDBSubscriptionClient
+}
+
+type TopicPubSubClient struct {
+	Client zindexer.ITopicPubSubClient
+}

connections/types.go

@@ -3,9 +3,12 @@ module github.com/Zondax/zindexer
 go 1.15
 
 require (
+	github.com/ThreeDotsLabs/watermill v1.1.1
+	github.com/hasura/go-graphql-client v0.2.0
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/viper v1.7.1
 	go.uber.org/zap v1.16.0
+	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
 	gorm.io/driver/postgres v1.0.6
 	gorm.io/gorm v1.20.11
 )