prepare for v1.1.0 release

Makefile

@@ -1,2 +1,4 @@
 test:
-	go test -v -cover ./graphql .
+	GO111MODULE=on go test -v -count=1 -cover ./graphql .
+
+.PHONY: test

client.go

@@ -1,33 +1,83 @@
 package appsync
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"log"
 	"net/http"
+	"time"
 
+	v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
 	"github.com/sony/appsync-client-go/graphql"
 )
 
 // Client is the AppSync GraphQL API client
 type Client struct {
 	graphQLAPI   GraphQLClient
 	subscriberID string
+	iamAuth      *struct {
+		signer v4.Signer
+		region string
+		host   string
+	}
 }
 
 // NewClient returns a Client instance.
 func NewClient(graphql GraphQLClient, opts ...ClientOption) *Client {
 	c := &Client{graphQLAPI: graphql}
 	for _, opt := range opts {
-		opt.Apply(c)
+		opt(c)
 	}
 	return c
 }
 
+func (c *Client) sleepIfNeeded(request graphql.PostRequest) {
+	if request.IsSubscription() {
+		// Here be dragons.
+		time.Sleep(2 * time.Second)
+	}
+}
+
+func (c *Client) signRequest(request graphql.PostRequest) (http.Header, error) {
+	jsonBytes, err := json.Marshal(request)
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+
+	req, err := http.NewRequest("POST", c.iamAuth.host, bytes.NewBuffer(jsonBytes))
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+
+	_, err = c.iamAuth.signer.Sign(req, bytes.NewReader(jsonBytes), "appsync", c.iamAuth.region, time.Now())
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	return req.Header, nil
+}
+
 //Post is a synchronous AppSync GraphQL POST request.
 func (c *Client) Post(request graphql.PostRequest) (*graphql.Response, error) {
+	defer c.sleepIfNeeded(request)
 	header := http.Header{}
 	if request.IsSubscription() && len(c.subscriberID) > 0 {
 		header.Set("x-amz-subscriber-id", c.subscriberID)
 	}
+
+	if c.iamAuth != nil {
+		h, err := c.signRequest(request)
+		if err != nil {
+			log.Println(err)
+			return nil, err
+		}
+		for k, v := range h {
+			header[k] = v
+		}
+	}
 	return c.graphQLAPI.Post(header, request)
 }
 
@@ -37,5 +87,19 @@ func (c *Client) PostAsync(request graphql.PostRequest, callback func(*graphql.R
 	if request.IsSubscription() && len(c.subscriberID) > 0 {
 		header.Set("x-amz-subscriber-id", c.subscriberID)
 	}
-	return c.graphQLAPI.PostAsync(header, request, callback)
+	cb := func(g *graphql.Response, err error) {
+		c.sleepIfNeeded(request)
+		callback(g, err)
+	}
+	if c.iamAuth != nil {
+		h, err := c.signRequest(request)
+		if err != nil {
+			log.Println(err)
+			return nil, err
+		}
+		for k, v := range h {
+			header[k] = v
+		}
+	}
+	return c.graphQLAPI.PostAsync(header, request, cb)
 }

extensions.go

@@ -16,9 +16,16 @@ type Extensions struct {
 			Topics []string `json:"topics"`
 			Client string   `json:"client"`
 		} `json:"mqttConnections"`
+		NewSubscriptions map[string]Subscription `json:"newSubscriptions"`
 	} `json:"subscription"`
 }
 
+// Subscription represents AWS AppSync subscription mqtt topic
+type Subscription struct {
+	Topic      string      `json:"topic"`
+	ExpireTime interface{} `json:"expireTime"`
+}
+
 // NewExtensions returns Extensions instance
 func NewExtensions(response *graphql.Response) (*Extensions, error) {
 	j, ok := (*response.Extensions).(map[string]interface{})

go.mod

@@ -3,11 +3,15 @@ module github.com/sony/appsync-client-go
 go 1.14
 
 require (
-	github.com/cenkalti/backoff v2.0.0+incompatible
+	github.com/aws/aws-sdk-go v1.28.11
+	github.com/cenkalti/backoff v2.1.1+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.1.1
 	github.com/google/uuid v1.1.1
 	github.com/gorilla/websocket v1.4.0
 	github.com/graph-gophers/graphql-go v0.0.0-20190225005345-3e8838d4614c
-	github.com/opentracing/opentracing-go v1.0.2
-	golang.org/x/net v0.0.0-20181220203305-927f97764cc3
+	github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 // indirect
+	github.com/k0kubun/pp v3.0.1+incompatible
+	github.com/mattn/go-colorable v0.1.4 // indirect
+	github.com/stretchr/testify v1.3.0 // indirect
+	golang.org/x/text v0.3.2 // indirect
 )

go.sum

@@ -1,5 +1,9 @@
-github.com/cenkalti/backoff v2.0.0+incompatible h1:5IIPUHhlnUZbcHQsQou5k1Tn58nJkeJL9U+ig5CHJbY=
-github.com/cenkalti/backoff v2.0.0+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
+github.com/aws/aws-sdk-go v1.28.11 h1:L2G5qI91s51cUP3hJli4mXRIZZ3alZHcwHWOJdMclKk=
+github.com/aws/aws-sdk-go v1.28.11/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
+github.com/cenkalti/backoff v2.1.1+incompatible h1:tKJnvO2kl0zmb/jA5UKAt4VoEVw1qxKWjE/Bpp46npY=
+github.com/cenkalti/backoff v2.1.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/eclipse/paho.mqtt.golang v1.1.1 h1:iPJYXJLaViCshRTW/PSqImSS6HJ2Rf671WR0bXZ2GIU=
 github.com/eclipse/paho.mqtt.golang v1.1.1/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
 github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
@@ -8,8 +12,27 @@ github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/graph-gophers/graphql-go v0.0.0-20190225005345-3e8838d4614c h1:YyFUsspLqAt3noyPCLz7EFK/o1LpC1j/6MjU0bSVOQ4=
 github.com/graph-gophers/graphql-go v0.0.0-20190225005345-3e8838d4614c/go.mod h1:uJhtPXrcJLqyi0H5IuMFh+fgW+8cMMakK3Txrbk/WJE=
+github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
+github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
+github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=
+github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
+github.com/k0kubun/pp v3.0.1+incompatible h1:3tqvf7QgUnZ5tXO6pNAZlrvHgl6DvifjDrd9g2S9Z40=
+github.com/k0kubun/pp v3.0.1+incompatible/go.mod h1:GWse8YhT0p8pT4ir3ZgBbfZild3tgzSScAn6HmfYukg=
+github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA=
+github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-isatty v0.0.8 h1:HLtExJ+uU2HOZ+wI0Tt5DtUDrx8yhUqDcp7fYERX4CE=
+github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/opentracing/opentracing-go v1.0.2 h1:3jA2P6O1F9UOrWVpwrIo17pu01KWvNWg4X946/Y5Zwg=
 github.com/opentracing/opentracing-go v1.0.2/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
-golang.org/x/net v0.0.0-20181029044818-c44066c5c816/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 golang.org/x/net v0.0.0-20181220203305-927f97764cc3 h1:eH6Eip3UpmR+yM/qI9Ijluzb1bNv/cAU/n+6l8tRSis=
 golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223 h1:DH4skfRX4EBpamg7iV4ZlCpblAHI6s6TDM39bFZumv8=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

graphql/client.go

@@ -4,22 +4,35 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"fmt"
-	"io/ioutil"
 	"log"
 	"net/http"
 	"net/url"
 	"time"
 
+	v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
 	"github.com/cenkalti/backoff"
 )
 
+type httpStatusError struct {
+	StatusCode int
+}
+
+func (e httpStatusError) Error() string {
+	return http.StatusText(e.StatusCode)
+}
+
+func (e httpStatusError) shouldRetry() bool {
+	return e.StatusCode == http.StatusInternalServerError ||
+		e.StatusCode == http.StatusServiceUnavailable
+}
+
 // Client represents a generic GraphQL API client.
 type Client struct {
 	endpoint       string
 	timeout        time.Duration
 	maxElapsedTime time.Duration
 	header         http.Header
+	signer         *v4.Signer
 }
 
 // NewClient returns a Client instance.
@@ -31,7 +44,7 @@ func NewClient(endpoint string, opts ...ClientOption) *Client {
 		header:         map[string][]string{},
 	}
 	for _, opt := range opts {
-		opt.Apply(c)
+		opt(c)
 	}
 
 	return c
@@ -65,7 +78,6 @@ func (c *Client) PostAsync(header http.Header, request PostRequest, callback fun
 		return nil, err
 	}
 
-	req.Header.Set("Content-Type", "application/json")
 	for k, v := range c.header {
 		req.Header[k] = v
 	}
@@ -83,61 +95,47 @@ func (c *Client) PostAsync(header http.Header, request PostRequest, callback fun
 		b.MaxElapsedTime = c.maxElapsedTime
 
 		var response Response
-		var errToCallback error
-		_ = backoff.Retry(func() error {
-			r, e := http.DefaultClient.Do(req)
-			if e != nil {
-				log.Println(e)
-				errToCallback = e
-				if e.(*url.Error).Timeout() {
-					if t, ok := http.DefaultTransport.(*http.Transport); ok {
-						t.CloseIdleConnections()
-					}
+		op := func() error {
+			r, err := http.DefaultClient.Do(req)
+			if err != nil {
+				log.Println(err)
+				if err.(*url.Error).Timeout() {
+					http.DefaultClient.CloseIdleConnections()
 				}
-				return nil
+				return backoff.Permanent(err)
 			}
 			defer func() {
-				if e := r.Body.Close(); e != nil {
-					log.Println(e)
+				if err := r.Body.Close(); err != nil {
+					log.Println(err)
 				}
 			}()
 
-			body, e := ioutil.ReadAll(r.Body)
-			if e != nil {
-				log.Println(e)
-				errToCallback = e
-				return nil
-			}
-
 			if r.StatusCode != http.StatusOK {
-				response.StatusCode = &r.StatusCode
-				errors := []interface{}{http.StatusText(r.StatusCode)}
-				response.Errors = &errors
-				// should retry
-				if r.StatusCode == http.StatusInternalServerError || r.StatusCode == http.StatusServiceUnavailable {
-					e := fmt.Errorf(http.StatusText(r.StatusCode))
-					log.Println(e)
-					return e
+				httpErr := httpStatusError{StatusCode: r.StatusCode}
+				if httpErr.shouldRetry() {
+					log.Println(httpErr)
+					return httpErr
 				}
-				return nil
+				return backoff.Permanent(httpErr)
 			}
 
-			if e := json.Unmarshal(body, &response); e != nil {
-				log.Println(e)
-				errToCallback = e
-				return nil
+			if err := json.NewDecoder(r.Body).Decode(&response); err != nil {
+				log.Println(err)
+				return backoff.Permanent(err)
 			}
 			response.StatusCode = &r.StatusCode
 
 			return nil
-		}, b)
-
-		if errToCallback != nil {
-			callback(nil, errToCallback)
-			return
 		}
 
-		callback(&response, nil)
+		switch err := backoff.Retry(op, b).(type) {
+		case nil:
+			callback(&response, nil)
+		case httpStatusError:
+			callback(&Response{&(err.StatusCode), nil, &[]interface{}{err.Error()}, nil}, nil)
+		default:
+			callback(nil, err)
+		}
 	}()
 
 	return cancel, nil

graphql/client_test.go

@@ -143,6 +143,48 @@ func TestInternalServerError(t *testing.T) {
 	}
 }
 
+func TestUnauthorizedError(t *testing.T) {
+	server := newUnauthorizedErrorServer()
+	defer server.Close()
+
+	unauthorizedError := http.StatusUnauthorized
+	errors = []interface{}{http.StatusText(unauthorizedError)}
+	want := Response{
+		StatusCode: &unauthorizedError,
+		Data:       nil,
+		Errors:     &errors,
+		Extensions: nil,
+	}
+
+	client := NewClient(server.URL, WithMaxElapsedTime(1*time.Microsecond))
+	got, err := client.Post(http.Header{}, PostRequest{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got == nil {
+		t.Fatal("got is nil")
+	}
+	checkResponse(t, &want, got)
+
+	resCh := make(chan *Response, 1)
+	errCh := make(chan error, 1)
+	cancel, err := client.PostAsync(http.Header{}, PostRequest{},
+		func(r *Response, err error) {
+			resCh <- r
+			errCh <- err
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cancel == nil {
+		t.Fatal("cancel is nil")
+	}
+	checkResponse(t, &want, <-resCh)
+	if err := <-errCh; err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestTimeout(t *testing.T) {
 	server := newDelayServer(3 * time.Microsecond)
 	defer server.Close()

graphql/graphql_test.go

@@ -37,6 +37,12 @@ func newInternalServerErrorServer() *httptest.Server {
 	}))
 }
 
+func newUnauthorizedErrorServer() *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, http.StatusText(http.StatusUnauthorized), http.StatusUnauthorized)
+	}))
+}
+
 func newDelayServer(delay time.Duration) *httptest.Server {
 	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		time.Sleep(delay)