tweak connect.Client

internal/ctl/connectors/add.go

@@ -61,7 +61,7 @@ func doAddConnectors(cmd *cobra.Command, params *addConnectorsCmdParams) error {
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/connectors/list.go

@@ -51,7 +51,7 @@ func doListConnectors(_ *cobra.Command, params *listConnectorsCmdParams) error {
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/connectors/manage.go

@@ -80,8 +80,7 @@ func doManageConnectors(cmd *cobra.Command, params *manageConnectorsCmdParams) e
 
 	clusterLogger.Debug("executing manage connectors command")
 
-	err := checkConfig(params)
-	if err != nil {
+	if err := checkConfig(params); err != nil {
 		return errors.Wrap(err, "Error with configuration")
 	}
 
@@ -103,7 +102,7 @@ func doManageConnectors(cmd *cobra.Command, params *manageConnectorsCmdParams) e
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}
@@ -113,20 +112,16 @@ func doManageConnectors(cmd *cobra.Command, params *manageConnectorsCmdParams) e
 		return errors.Wrap(err, "Error creating connectors manager")
 	}
 
-	ctx := context.Background()
-
 	if params.EnableHealthCheck {
 		healthCheckHandler := healthcheck.New(mngr)
 
 		go func() {
-			err := healthCheckHandler.Start(params.HealthCheckAddress)
-			if err != nil {
+			if err := healthCheckHandler.Start(params.HealthCheckAddress); err != nil {
 				clusterLogger.WithError(err).Fatalln("Error starting healthcheck")
 			}
 		}()
-
 		// nolint
-		defer healthCheckHandler.Shutdown(ctx)
+		defer healthCheckHandler.Shutdown(context.Background())
 	}
 
 	if params.RunOnce {
@@ -135,13 +130,10 @@ func doManageConnectors(cmd *cobra.Command, params *manageConnectorsCmdParams) e
 		}
 	} else {
 		stopCh := signals.SetupSignalHandler()
-
 		if err := mngr.Manage(source, stopCh); err != nil {
 			return errors.Wrap(err, "Error running connector manager")
 		}
 	}
-
-	clusterLogger.Info("finished executing manage connectors command")
 	return nil
 }
 

internal/ctl/connectors/pause.go

@@ -48,7 +48,7 @@ func doPauseConnectors(_ *cobra.Command, params *pauseConnectorsCmdParams) error
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/connectors/remove.go

@@ -48,7 +48,7 @@ func doRemoveConnectors(_ *cobra.Command, params *removeConnectorsCmdParams) err
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/connectors/restart.go

@@ -62,7 +62,7 @@ func doRestartConnectors(_ *cobra.Command, params *restartConnectorsCmdParams) e
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/connectors/resume.go

@@ -48,7 +48,7 @@ func doResumeConnectors(_ *cobra.Command, params *resumeConnectorsCmdParams) err
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

internal/ctl/plugins/list.go

@@ -51,7 +51,7 @@ func doListPlugins(_ *cobra.Command, params *listPluginsCmdParams) error {
 
 	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
 
-	client, err := connect.NewClient(params.ClusterURL, userAgent)
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
 	if err != nil {
 		return errors.Wrap(err, "error creating connect client")
 	}

pkg/client/connect/client.go

@@ -30,31 +30,47 @@ type Client struct {
 
 	// HTTP client used to communicate with the API. By default
 	// http.DefaultClient will be used.
-	HTTPClient *http.Client
+	httpClient *http.Client
 
 	// User agent used when communicating with the Kafka Connect API.
-	UserAgent string
+	userAgent string
+}
+
+// Option can be supplied that override the default Clients properties
+type Option func(c *Client)
+
+// WithUserAgent allows the userAgent to be overridden
+func WithUserAgent(userAgent string) Option {
+	return func(c *Client) {
+		c.userAgent = userAgent
+	}
+}
+
+// WithHTTPClient allows a specific http.Client to be set
+func WithHTTPClient(httpClient *http.Client) Option {
+	return func(c *Client) {
+		c.httpClient = httpClient
+	}
 }
 
 // NewClient returns a new Kafka Connect API client that communicates host.
-func NewClient(host string, userAgent string) (*Client, error) {
+func NewClient(host string, opts ...Option) (*Client, error) {
 	hostURL, err := url.Parse(host)
 	if err != nil {
 		return nil, errors.Wrapf(err, "parsing url %s", host)
 	}
 
-	if userAgent == "" {
-		userAgent = userAgentDefault
+	c := &Client{
+		host:       hostURL,
+		userAgent:  userAgentDefault,
+		httpClient: http.DefaultClient,
 	}
 
-	return &Client{host: hostURL, UserAgent: userAgent}, nil
-}
-
-func (c *Client) httpClient() *http.Client {
-	if c.HTTPClient == nil {
-		return http.DefaultClient
+	for _, opt := range opts {
+		opt(c)
 	}
-	return c.HTTPClient
+
+	return c, nil
 }
 
 // Host returns the API root URL the Client is configured to talk to.
@@ -95,9 +111,7 @@ func (c *Client) NewRequest(method, path string, body interface{}) (*http.Reques
 	if contentType != "" {
 		request.Header.Set("Content-Type", contentType)
 	}
-	if c.UserAgent != "" {
-		request.Header.Set("User-Agent", c.UserAgent)
-	}
+	request.Header.Set("User-Agent", c.userAgent)
 
 	return request, nil
 }
@@ -106,7 +120,7 @@ func (c *Client) NewRequest(method, path string, body interface{}) (*http.Reques
 // JSON-decoded and stored in the value pointed to by v, or returned as an
 // error if an API or HTTP error has occurred.
 func (c *Client) Do(req *http.Request, v interface{}) (*http.Response, error) {
-	response, err := c.httpClient().Do(req)
+	response, err := c.httpClient.Do(req)
 	if err != nil {
 		return nil, err
 	}

pkg/manager/manager.go

@@ -36,12 +36,6 @@ type ConnectorManager struct {
 
 // NewConnectorsManager creates a new ConnectorManager
 func NewConnectorsManager(client client, config *Config) (*ConnectorManager, error) {
-	//userAgent := fmt.Sprintf("90poe.io/connectctl/%s", config.Version)
-
-	//client, err := connect.NewClient(config.ClusterURL, userAgent)
-	//if err != nil {
-	//	return nil, errors.Wrap(err, "error creating connect client")
-	//}
 	return &ConnectorManager{
 		config:         config,
 		client:         client,