add CredentialFetcher to allow updating credentials on each newOption…

…s creation

client.go

@@ -167,9 +167,9 @@ func toClientOptions(c *Client, o *clientOptions) *mqtt.ClientOptions {
 		opts.SetClientID(o.clientID)
 	}
 
+	setCredentials(o, opts)
+
 	opts.AddBroker(formatAddressWithProtocol(o)).
-		SetUsername(o.username).
-		SetPassword(o.password).
 		SetTLSConfig(o.tlsConfig).
 		SetAutoReconnect(o.autoReconnect).
 		SetCleanSession(o.cleanSession).
@@ -184,6 +184,23 @@ func toClientOptions(c *Client, o *clientOptions) *mqtt.ClientOptions {
 	return opts
 }
 
+func setCredentials(o *clientOptions, opts *mqtt.ClientOptions) {
+	if o.credentialFetcher != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), o.credentialFetchTimeout)
+		defer cancel()
+
+		if c, err := o.credentialFetcher.Credentials(ctx); err == nil {
+			opts.SetUsername(c.Username)
+			opts.SetPassword(c.Password)
+
+			return
+		}
+	}
+
+	opts.SetUsername(o.username)
+	opts.SetPassword(o.password)
+}
+
 func formatAddressWithProtocol(opts *clientOptions) string {
 	if opts.tlsConfig != nil {
 		return fmt.Sprintf("tls://%s", opts.brokerAddress)

client_credentials.go

@@ -0,0 +1,21 @@
+package courier
+
+import "context"
+
+// Credential is a <username,password> pair.
+type Credential struct {
+	Username string
+	Password string
+}
+
+// CredentialFetcher is an interface that allows to fetch credentials for a client.
+type CredentialFetcher interface {
+	Credentials(context.Context) (*Credential, error)
+}
+
+// WithCredentialFetcher sets the specified CredentialFetcher.
+func WithCredentialFetcher(fetcher CredentialFetcher) ClientOption {
+	return optionFunc(func(o *clientOptions) {
+		o.credentialFetcher = fetcher
+	})
+}

client_credentials_test.go

@@ -0,0 +1,26 @@
+package courier
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type mockCredentialFetcher struct {
+	mock.Mock
+}
+
+func newMockCredentialFetcher(t *testing.T) *mockCredentialFetcher {
+	m := &mockCredentialFetcher{}
+	m.Test(t)
+	return m
+}
+
+func (m *mockCredentialFetcher) Credentials(ctx context.Context) (*Credential, error) {
+	args := m.Called(ctx)
+	if c := args.Get(0); c != nil {
+		return c.(*Credential), args.Error(1)
+	}
+	return nil, args.Error(1)
+}

client_options.go

@@ -197,14 +197,16 @@ func WithExponentialStartOptions(options ...StartOption) ClientOption {
 type clientOptions struct {
 	username, clientID, password,
 	brokerAddress string
-	resolver Resolver
+	resolver          Resolver
+	credentialFetcher CredentialFetcher
 
 	tlsConfig *tls.Config
 
 	autoReconnect, maintainOrder, cleanSession bool
 
 	connectTimeout, writeTimeout, keepAlive,
-	maxReconnectInterval, gracefulShutdownPeriod time.Duration
+	maxReconnectInterval, gracefulShutdownPeriod,
+	credentialFetchTimeout time.Duration
 
 	startOptions *startOptions
 
@@ -230,6 +232,7 @@ func defaultClientOptions() *clientOptions {
 		maxReconnectInterval:   5 * time.Minute,
 		gracefulShutdownPeriod: 30 * time.Second,
 		keepAlive:              60 * time.Second,
+		credentialFetchTimeout: 10 * time.Second,
 		newEncoder:             DefaultEncoderFunc,
 		newDecoder:             DefaultDecoderFunc,
 		store:                  inMemoryPersistence,

client_options_test.go

@@ -20,6 +20,7 @@ func TestClientOptionSuite(t *testing.T) {
 func (s *ClientOptionSuite) Test_apply() {
 	store := NewMemoryStore()
 	r := resolver{}
+	mc := newMockCredentialFetcher(s.T())
 
 	tests := []struct {
 		name   string
@@ -101,6 +102,11 @@ func (s *ClientOptionSuite) Test_apply() {
 			option: WithResolver(r),
 			want:   &clientOptions{resolver: r},
 		},
+		{
+			name:   "WithCredentialFetcher",
+			option: WithCredentialFetcher(mc),
+			want:   &clientOptions{credentialFetcher: mc},
+		},
 		{
 			name:   "WithExponentialStartOptions",
 			option: WithExponentialStartOptions(WithMaxInterval(time.Second)),
@@ -192,6 +198,7 @@ func (s *ClientOptionSuite) Test_defaultOptions() {
 		maxReconnectInterval:   5 * time.Minute,
 		gracefulShutdownPeriod: 30 * time.Second,
 		keepAlive:              60 * time.Second,
+		credentialFetchTimeout: 10 * time.Second,
 		newEncoder:             DefaultEncoderFunc,
 		newDecoder:             DefaultDecoderFunc,
 		store:                  inMemoryPersistence,

client_test.go

@@ -196,6 +196,37 @@ func TestNewClientWithResolverOption(t *testing.T) {
 	mr.AssertExpectations(t)
 }
 
+func TestNewClientWithCredentialFetcher(t *testing.T) {
+	mcf := newMockCredentialFetcher(t)
+	mcf.On("Credentials", mock.Anything).Return(&Credential{
+		Username: "username",
+		Password: "password",
+	}, nil)
+
+	newClientFunc.Store(func(opts *mqtt.ClientOptions) mqtt.Client {
+		assert.Equal(t, "username", opts.Username)
+		assert.Equal(t, "password", opts.Password)
+		return mqtt.NewClient(opts)
+	})
+	defer func() {
+		newClientFunc.Store(mqtt.NewClient)
+	}()
+
+	c, err := NewClient(append(defOpts, WithCredentialFetcher(mcf))...)
+
+	assert.NoError(t, c.Start())
+	mcf.AssertExpectations(t)
+
+	assert.Eventually(t, func() bool {
+		return c.IsConnected()
+	}, 10*time.Second, 250*time.Millisecond)
+
+	c.Stop()
+
+	assert.NoError(t, err)
+	mcf.AssertExpectations(t)
+}
+
 func TestNewClientWithExponentialStartOptions(t *testing.T) {
 	c, err := NewClient(append(defOpts, WithExponentialStartOptions(WithMaxInterval(10*time.Second)))...)
 	assert.NoError(t, err)

docs/docs/sdk/SDK.md

@@ -33,6 +33,7 @@ Package courier contains the client that can be used to interact with the courie
   - [func WithCleanSession\(cleanSession bool\) ClientOption](#WithCleanSession)
   - [func WithClientID\(clientID string\) ClientOption](#WithClientID)
   - [func WithConnectTimeout\(duration time.Duration\) ClientOption](#WithConnectTimeout)
+  - [func WithCredentialFetcher\(fetcher CredentialFetcher\) ClientOption](#WithCredentialFetcher)
   - [func WithCustomDecoder\(decoderFunc DecoderFunc\) ClientOption](#WithCustomDecoder)
   - [func WithCustomEncoder\(encoderFunc EncoderFunc\) ClientOption](#WithCustomEncoder)
   - [func WithExponentialStartOptions\(options ...StartOption\) ClientOption](#WithExponentialStartOptions)
@@ -52,6 +53,8 @@ Package courier contains the client that can be used to interact with the courie
   - [func WithUsername\(username string\) ClientOption](#WithUsername)
   - [func WithWriteTimeout\(duration time.Duration\) ClientOption](#WithWriteTimeout)
 - [type ConnectionInformer](#ConnectionInformer)
+- [type Credential](#Credential)
+- [type CredentialFetcher](#CredentialFetcher)
 - [type Decoder](#Decoder)
   - [func DefaultDecoderFunc\(\_ context.Context, r io.Reader\) Decoder](#DefaultDecoderFunc)
 - [type DecoderFunc](#DecoderFunc)
@@ -366,6 +369,15 @@ func WithConnectTimeout(duration time.Duration) ClientOption
 
 WithConnectTimeout limits how long the client will wait when trying to open a connection to an MQTT server before timing out. A duration of 0 never times out. Default 15 seconds.
 
+<a name="WithCredentialFetcher"></a>
+### func [WithCredentialFetcher](https://github.com/gojek/courier-go/blob/main/client_credentials.go#L17)
+
+```go
+func WithCredentialFetcher(fetcher CredentialFetcher) ClientOption
+```
+
+WithCredentialFetcher sets the specified CredentialFetcher.
+
 <a name="WithCustomDecoder"></a>
 ### func [WithCustomDecoder](https://github.com/gojek/courier-go/blob/main/client_options.go#L176)
 
@@ -542,6 +554,29 @@ type ConnectionInformer interface {
 }
 ```
 
+<a name="Credential"></a>
+## type [Credential](https://github.com/gojek/courier-go/blob/main/client_credentials.go#L6-L9)
+
+Credential is a \<username,password\> pair.
+
+```go
+type Credential struct {
+    Username string
+    Password string
+}
+```
+
+<a name="CredentialFetcher"></a>
+## type [CredentialFetcher](https://github.com/gojek/courier-go/blob/main/client_credentials.go#L12-L14)
+
+CredentialFetcher is an interface that allows to fetch credentials for a client.
+
+```go
+type CredentialFetcher interface {
+    Credentials(context.Context) (*Credential, error)
+}
+```
+
 <a name="Decoder"></a>
 ## type [Decoder](https://github.com/gojek/courier-go/blob/main/decoder.go#L16-L19)