Repository: ably/ably-go
Pull request #196 (2020-09-07 13:30:55)

Merge main into 1.2
Pull request just for CI; won't wait for review as this is just a merge plus minor porting.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7bbd9ac..2cfa4c6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,15 @@
 # Change Log
 
+## [v1.1.5](https://github.com/ably/ably-go/tree/v1.1.5) (2020-05-27)
+
+[Full Changelog](https://github.com/ably/ably-go/compare/v1.1.4...v1.1.5)
+
+**Merged pull requests:**
+
+- Add logs to help in troubleshooting RestClient http api calls [\#162](https://github.com/ably/ably-go/pull/162) ([gernest](https://github.com/gernest))
+- Remove unused fields definition on RealtimeClient [\#161](https://github.com/ably/ably-go/pull/161) ([gernest](https://github.com/gernest))
+- add implementation for  RTN15b-c spec [\#159](https://github.com/ably/ably-go/pull/159) ([gernest](https://github.com/gernest))
+
 ## [v1.1.4](https://github.com/ably/ably-go/tree/v1.1.4) (2020-04-24)
 
 [Full Changelog](https://github.com/ably/ably-go/compare/v1.1.3...v1.1.4)
diff --git a/README.md b/README.md
index 0391821..be42391 100644
--- a/README.md
+++ b/README.md
@@ -268,7 +268,7 @@ As the library is actively developed couple of features are not there yet:
 
 This library uses [semantic versioning](http://semver.org/). For each release, the following needs to be done:
 
-* Create a branch for the release, named like `release/1.1.5`
+* Create a branch for the release, named like `release/1.1.6`
 * Replace all references of the current version number with the new version number and commit the changes
 * Run [`github_changelog_generator`](https://github.com/github-changelog-generator/github-changelog-generator) to automate the update of the [CHANGELOG](./CHANGELOG.md). This may require some manual intervention, both in terms of how the command is run and how the change log file is modified. Your mileage may vary:
     * The command you will need to run will look something like this: `github_changelog_generator -u ably -p ably-go --since-tag v1.1.4 --output delta.md`
@@ -277,9 +277,9 @@ This library uses [semantic versioning](http://semver.org/). For each release, t
     * Also ensure that the "Full Changelog" link points to the new version tag instead of the `HEAD`
     * Commit this change: `git add CHANGELOG.md && git commit -m "Update change log."`
 * Commit [CHANGELOG](./CHANGELOG.md)
-* Make a PR against `master`
-* Once the PR is approved, merge it into `master`
-* Add a tag to the new `master` head commit and push to origin such as `git tag v1.1.5 && git push origin v1.1.5`
+* Make a PR against `main`
+* Once the PR is approved, merge it into `main`
+* Add a tag to the new `main` head commit and push to origin such as `git tag v1.1.6 && git push origin v1.1.6`
 
 ## Support and feedback
 
diff --git a/ably/auth.go b/ably/auth.go
index 50a80de..6701758 100644
--- a/ably/auth.go
+++ b/ably/auth.go
@@ -180,10 +180,13 @@ func (a *Auth) RequestToken(params *TokenParams, opts AuthOptions) (*TokenDetail
 }
 
 func (a *Auth) requestToken(params *TokenParams, opts *authOptions) (tok *TokenDetails, tokReqClientID string, err error) {
+	log := a.logger().Sugar()
 	switch {
 	case opts != nil && opts.Token != "":
+		log.Verbose("Auth: found token in AuthOptions")
 		return newTokenDetails(opts.Token), "", nil
 	case opts != nil && opts.TokenDetails != nil:
+		log.Verbose("Auth: found TokenDetails in AuthOptions")
 		return opts.TokenDetails, "", nil
 	}
 	if params == nil {
@@ -193,8 +196,10 @@ func (a *Auth) requestToken(params *TokenParams, opts *authOptions) (tok *TokenD
 	var tokReq *TokenRequest
 	switch {
 	case opts.AuthCallback != nil:
+		log.Verbose("Auth: found AuthCallback in AuthOptions")
 		v, err := opts.AuthCallback(context.TODO(), *params)
 		if err != nil {
+			log.Error("Auth: failed calling opts.AuthCallback ", err)
 			return nil, "", newError(ErrErrorFromClientTokenCallback, err)
 		}
 
@@ -218,8 +223,10 @@ func (a *Auth) requestToken(params *TokenParams, opts *authOptions) (tok *TokenD
 			panic(fmt.Errorf("unhandled TokenLike: %T", v))
 		}
 	case opts.AuthURL != "":
+		log.Verbose("Auth: found AuthURL in AuthOptions")
 		res, err := a.requestAuthURL(params, opts)
 		if err != nil {
+			log.Error("Auth: failed calling requesting token with AuthURL ", err)
 			return nil, "", err
 		}
 		switch res := res.(type) {
@@ -230,6 +237,8 @@ func (a *Auth) requestToken(params *TokenParams, opts *authOptions) (tok *TokenD
 			tokReqClientID = tokReq.ClientID
 		}
 	default:
+		log.Verbose("Auth: using default token request")
+
 		req, err := a.createTokenRequest(params, opts)
 		if err != nil {
 			return nil, "", err
@@ -277,6 +286,7 @@ func (a *Auth) Authorize(params *TokenParams, setOpts AuthOptions) (*TokenDetail
 }
 
 func (a *Auth) authorize(params *TokenParams, opts *authOptions, force bool) (*TokenDetails, error) {
+	log := a.logger().Sugar()
 	switch tok := a.token(); {
 	case tok != nil && !force && (tok.Expires == 0 || !tok.expired(a.opts().Now())):
 		return tok, nil
@@ -285,18 +295,22 @@ func (a *Auth) authorize(params *TokenParams, opts *authOptions, force bool) (*T
 	case params == nil && a.clientID != "":
 		params = &TokenParams{ClientID: a.clientID}
 	}
+	log.Info("Auth: sending  token request")
 	tok, tokReqClientID, err := a.requestToken(params, opts)
 	if err != nil {
+		log.Error("Auth: failed to get token", err)
 		return nil, err
 	}
 	// Fail if the non-empty ClientID, that was set explicitely via clientOptions, does
 	// not match the non-wildcard ClientID returned with the token.
 	if areClientIDsSet(a.clientID, tok.ClientID) && a.clientID != tok.ClientID {
+		log.Error("Auth: ", errClientIDMismatch)
 		return nil, newError(ErrInvalidClientID, errClientIDMismatch)
 	}
 	// Fail if non-empty ClientID requested by a TokenRequest
 	// does not match the non-wildcard ClientID that arrived with the token.
 	if areClientIDsSet(tokReqClientID, tok.ClientID) && tokReqClientID != tok.ClientID {
+		log.Error("Auth: ", errClientIDMismatch)
 		return nil, newError(ErrInvalidClientID, errClientIDMismatch)
 	}
 	a.method = authToken
@@ -309,6 +323,7 @@ func (a *Auth) authorize(params *TokenParams, opts *authOptions, force bool) (*T
 func (a *Auth) reauthorize() (*TokenDetails, error) {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
+	a.logger().Sugar().Info("Auth: reauthorize")
 	return a.authorize(a.params, nil, true)
 }
 
diff --git a/ably/logger.go b/ably/logger.go
index c4e720b..3f96aa5 100644
--- a/ably/logger.go
+++ b/ably/logger.go
@@ -61,6 +61,10 @@ func (l LoggerOptions) GetLogger() Logger {
 	return defaultLog.Logger
 }
 
+func (l LoggerOptions) Sugar() *SugaredLogger {
+	return &SugaredLogger{LoggerOptions: l}
+}
+
 // Logger is an interface for ably loggers.
 type Logger interface {
 	Print(level LogLevel, v ...interface{})
@@ -82,3 +86,47 @@ func (s *StdLogger) Print(level LogLevel, v ...interface{}) {
 		s.Logger.Print(v...)
 	}
 }
+
+type SugaredLogger struct {
+	LoggerOptions
+}
+
+func (s SugaredLogger) Error(v ...interface{}) {
+	s.Print(LogError, v...)
+}
+
+func (s SugaredLogger) Errorf(fmt string, v ...interface{}) {
+	s.Printf(LogError, fmt, v...)
+}
+
+func (s SugaredLogger) Warn(v ...interface{}) {
+	s.Print(LogWarning, v...)
+}
+
+func (s SugaredLogger) Warnf(fmt string, v ...interface{}) {
+	s.Printf(LogWarning, fmt, v...)
+}
+
+func (s SugaredLogger) Info(v ...interface{}) {
+	s.Print(LogInfo, v...)
+}
+
+func (s SugaredLogger) Infof(fmt string, v ...interface{}) {
+	s.Printf(LogInfo, fmt, v...)
+}
+
+func (s SugaredLogger) Verbose(v ...interface{}) {
+	s.Print(LogVerbose, v...)
+}
+
+func (s SugaredLogger) Verbosef(fmt string, v ...interface{}) {
+	s.Printf(LogVerbose, fmt, v...)
+}
+
+func (s SugaredLogger) Debugf(fmt string, v ...interface{}) {
+	s.Printf(LogDebug, fmt, v...)
+}
+
+func (s SugaredLogger) Debug(v ...interface{}) {
+	s.Print(LogDebug, v...)
+}
diff --git a/ably/options.go b/ably/options.go
index a7ba6ee..f7b55e3 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -26,6 +26,7 @@ const (
 var defaultOptions = clientOptions{
 	RestHost:                 RestHost,
 	HTTPMaxRetryCount:        3,
+	HTTPRequestTimeout:       10 * time.Second,
 	RealtimeHost:             "realtime.ably.io",
 	TimeoutDisconnect:        30 * time.Second,
 	RealtimeRequestTimeout:   10 * time.Second, // DF1b
@@ -174,6 +175,10 @@ type clientOptions struct {
 
 	// max number of fallback hosts to use as a fallback.
 	HTTPMaxRetryCount int
+	// HTTPRequestTimeout is the timeout for getting a response for outgoing HTTP requests.
+	//
+	// Will only be used if no custom HTTPClient is set.
+	HTTPRequestTimeout time.Duration
 
 	// The period in milliseconds before HTTP requests are retried against the
 	// default endpoint
@@ -214,7 +219,7 @@ type clientOptions struct {
 
 	// HTTPClient specifies the client used for HTTP communication by RestClient.
 	//
-	// If HTTPClient is nil, the http.DefaultClient is used.
+	// If HTTPClient is nil, a client configured with default settings is used.
 	HTTPClient *http.Client
 
 	//When provided this will be used on every request.
@@ -307,7 +312,9 @@ func (opts *clientOptions) httpclient() *http.Client {
 	if opts.HTTPClient != nil {
 		return opts.HTTPClient
 	}
-	return http.DefaultClient
+	return &http.Client{
+		Timeout: defaultOptions.HTTPRequestTimeout,
+	}
 }
 
 func (opts *clientOptions) protocol() string {
diff --git a/ably/paginated_result.go b/ably/paginated_result.go
index 7595c5e..8579ef4 100644
--- a/ably/paginated_result.go
+++ b/ably/paginated_result.go
@@ -1,6 +1,9 @@
 package ably
 
 import (
+	"errors"
+	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"path"
@@ -135,9 +138,13 @@ func newPaginatedResult(opts *proto.ChannelOptions, req paginatedRequest) (*Pagi
 			return nil, err
 		}
 		p.errorCode = i
+	} else if !p.success {
+		return nil, malformedPaginatedResponseError(resp)
 	}
 	if h := p.respHeaders.Get(AblyErrormessageHeader); h != "" {
 		p.errorMessage = h
+	} else if !p.success {
+		return nil, malformedPaginatedResponseError(resp)
 	}
 	p.path = builtPath
 	p.links = resp.Header["Link"]
@@ -149,6 +156,17 @@ func newPaginatedResult(opts *proto.ChannelOptions, req paginatedRequest) (*Pagi
 	return p, nil
 }
 
+func malformedPaginatedResponseError(resp *http.Response) error {
+	body := make([]byte, 200)
+	n, err := io.ReadFull(resp.Body, body)
+	body = body[:n]
+	msg := fmt.Sprintf("invalid PaginatedResult HTTP response; status: %d; body (first %d bytes): %q", resp.StatusCode, len(body), body)
+	if err != nil && !errors.Is(err, io.EOF) {
+		return fmt.Errorf("%s; body read error: %w", msg, err)
+	}
+	return errors.New(msg)
+}
+
 func copyHeader(dest, src http.Header) {
 	for k, v := range src {
 		d := make([]string, len(v))
diff --git a/ably/rest_client.go b/ably/rest_client.go
index 5f9c217..89de5d6 100644
--- a/ably/rest_client.go
+++ b/ably/rest_client.go
@@ -13,6 +13,7 @@ import (
 	"mime"
 	"net/http"
 	"net/http/httptrace"
+	"net/http/httputil"
 	"reflect"
 	"strings"
 	"sync"
@@ -35,7 +36,7 @@ const (
 	AblyLibHeader          = "X-Ably-Lib"
 	AblyErrorCodeHeader    = "X-Ably-Errorcode"
 	AblyErrormessageHeader = "X-Ably-Errormessage"
-	LibraryVersion         = "1.1.4"
+	LibraryVersion         = "1.1.5"
 	LibraryName            = "ably-go"
 	LibraryString          = LibraryName + "-" + LibraryVersion
 	AblyVersion            = "1.0"
@@ -312,10 +313,12 @@ func (f *fallbackCache) put(host string) {
 }
 
 func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{}) (*http.Response, error)) (*http.Response, error) {
+	log := c.opts.Logger.Sugar()
 	if c.successFallbackHost == nil {
 		c.successFallbackHost = &fallbackCache{
 			duration: c.opts.fallbackRetryTimeout(),
 		}
+		log.Verbosef("RestClient: setup fallback duration to %v", c.successFallbackHost.duration)
 	}
 	req, err := c.NewHTTPRequest(r)
 	if err != nil {
@@ -323,16 +326,38 @@ func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{})
 	}
 	if h := c.successFallbackHost.get(); h != "" {
 		req.URL.Host = h // RSC15f
+		log.Verbosef("RestClient: setting URL.Host=%q", h)
 	}
 	if c.opts.Trace != nil {
 		req = req.WithContext(httptrace.WithClientTrace(req.Context(), c.opts.Trace))
+		log.Verbose("RestClient: enabling httptrace")
+	}
+	if log.Is(LogVerbose) {
+		b, err := httputil.DumpRequest(req, false)
+		if err != nil {
+			log.Error("RestClient: error trying to dump request: ", err)
+		} else {
+			log.Verbose("RestClient: ", string(b))
+		}
 	}
 	resp, err := c.opts.httpclient().Do(req)
 	if err != nil {
+		log.Error("RestClient: failed sending a request ", err)
 		return nil, newError(ErrInternalError, err)
 	}
+	if log.Is(LogVerbose) {
+		typ, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type"))
+		// dumping msgpack body isn't that helpbul when debugging
+		b, err := httputil.DumpResponse(resp, typ != "application/x-msgpack")
+		if err != nil {
+			log.Error("RestClient: error trying to dump response: ", err)
+		} else {
+			log.Verbose("RestClient: ", string(b))
+		}
+	}
 	resp, err = handle(resp, r.Out)
 	if err != nil {
+		log.Error("RestClient: error handling response: ", err)
 		if e, ok := err.(*ErrorInfo); ok {
 			if canFallBack(e.StatusCode) &&
 				(strings.HasPrefix(req.URL.Host, defaultOptions.RestHost) ||
@@ -341,6 +366,7 @@ func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{})
 				if c.opts.FallbackHosts != nil {
 					fallback = c.opts.FallbackHosts
 				}
+				log.Info("RestClient: trying to fallback with hosts=%v", fallback)
 				if len(fallback) > 0 {
 					left := fallback
 					iteration := 0
@@ -348,8 +374,11 @@ func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{})
 					if maxLimit == 0 {
 						maxLimit = defaultOptions.HTTPMaxRetryCount
 					}
+					log.Infof("RestClient: maximum fallback retry limit=%d", maxLimit)
+
 					for {
 						if len(left) == 0 {
+							log.Errorf("RestClient: exhausted fallback hosts", err)
 							return nil, err
 						}
 						var h string
@@ -369,15 +398,26 @@ func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{})
 						if err != nil {
 							return nil, err
 						}
+						log.Infof("RestClient:  chose fallback host=%q ", h)
 						req.URL.Host = h
 						req.Host = ""
 						req.Header.Set(HostHeader, h)
+						if log.Is(LogVerbose) {
+							b, err := httputil.DumpRequest(req, true)
+							if err != nil {
+								log.Error("RestClient: error trying to dump retry request with fallback host: ", err)
+							} else {
+								log.Verbose("RestClient: ", string(b))
+							}
+						}
 						resp, err := c.opts.httpclient().Do(req)
 						if err != nil {
+							log.Error("RestClient: failed sending a request to a fallback host", err)
 							return nil, newError(ErrInternalError, err)
 						}
 						resp, err = handle(resp, r.Out)
 						if err != nil {
+							log.Error("RestClient: error handling response: ", err)
 							if iteration == maxLimit-1 {
 								return nil, err
 							}
@@ -389,6 +429,16 @@ func (c *REST) doWithHandle(r *Request, handle func(*http.Response, interface{})
 							}
 							return nil, err
 						}
+						if log.Is(LogVerbose) {
+							typ, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type"))
+							// dumping msgpack body isn't that helpbul when debugging
+							b, err := httputil.DumpResponse(resp, typ != "application/x-msgpack")
+							if err != nil {
+								log.Error("RestClient: error trying to dump retry response: ", err)
+							} else {
+								log.Verbose("RestClient:: ", string(b))
+							}
+						}
 						c.successFallbackHost.put(h)
 						return resp, nil
 					}
@@ -457,13 +507,18 @@ func (c *REST) NewHTTPRequest(r *Request) (*http.Request, error) {
 }
 
 func (c *REST) handleResponse(resp *http.Response, out interface{}) (*http.Response, error) {
+	log := c.opts.Logger.Sugar()
+	log.Info("RestClient:checking valid http response")
 	if err := checkValidHTTPResponse(resp); err != nil {
+		log.Error("RestClient: failed to check valid http response ", err)
 		return nil, err
 	}
 	if out == nil {
 		return resp, nil
 	}
+	log.Info("RestClient: decoding response")
 	if err := decodeResp(resp, out); err != nil {
+		log.Error("RestClient: failed to decode response ", err)
 		return nil, err
 	}
 	return resp, nil


Test patch:
diff --git a/ably/paginated_result_test.go b/ably/paginated_result_test.go
index 6cd658d..50c07d0 100644
--- a/ably/paginated_result_test.go
+++ b/ably/paginated_result_test.go
@@ -1,6 +1,12 @@
 package ably_test
 
 import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/ably/ably-go/ably"
@@ -15,3 +21,38 @@ func TestPaginatedResult(t *testing.T) {
 		t.Errorf("expected %s got %s", expected, newPath)
 	}
 }
+
+func TestMalformedPaginatedResult(t *testing.T) {
+	bodyBytes, _ := json.Marshal([]string{"\x00 not really a PaginatedResult"})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(400)
+		w.Write(bodyBytes)
+	}))
+	defer srv.Close()
+
+	srvAddr := srv.Listener.Addr().(*net.TCPAddr)
+	client, err := ably.NewREST(ably.ClientOptions{}.
+		Token("xxxxxxx.yyyyyyy:zzzzzzz").
+		TLS(false).
+		RESTHost(srvAddr.IP.String()).
+		Port(srvAddr.Port))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := client.Request("POST", "/foo", nil, nil, nil)
+	if resp != nil {
+		t.Errorf("expected no HTTPPaginatedResult; got %+v", resp)
+	}
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	errMsg := err.Error()
+	if !strings.Contains(errMsg, "status: 400") {
+		t.Errorf("expected error to contain status code; got: %v", err)
+	}
+	if !strings.Contains(errMsg, fmt.Sprintf("%q", bodyBytes)) {
+		t.Errorf("expected error to contain body; got: %v", err)
+	}
+}