Repository: open-telemetry/opamp-go
Pull request #67 (2022-05-11 18:43:33)

Refactor to deduplicate common client functionality
- There are no changes in functionality. Tests are not touched (except package
  renaming).
- The OpAMP client logic that is common between WebSocket and plain HTTP
  transports is now implemented in one place: ClientCommon struct.
- Renamed HTTPLooper to HTTPSender to have more consistent naming (we have WSSender).
- StartSettings moved to types package (forced change to avoid dependency loop).


Patch:
diff --git a/client/client.go b/client/client.go
index 789191b..266512b 100644
--- a/client/client.go
+++ b/client/client.go
@@ -2,48 +2,11 @@ package client
 
 import (
 	"context"
-	"crypto/tls"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-type StartSettings struct {
-	// Connection parameters.
-
-	// Server URL. MUST be set.
-	OpAMPServerURL string
-
-	// Optional value of "Authorization" HTTP header in the HTTP request.
-	AuthorizationHeader string
-
-	// Optional TLS config for HTTP connection.
-	TLSConfig *tls.Config
-
-	// Agent information.
-	InstanceUid string
-
-	// AgentDescription MUST be set and MUST describe the Agent. See OpAMP spec
-	// for details.
-	AgentDescription *protobufs.AgentDescription
-
-	// Callbacks that the client will call after Start() returns nil.
-	Callbacks types.Callbacks
-
-	// Previously saved state. These will be reported to the server immediately
-	// after the connection is established.
-
-	// The hash of the last received remote config. If nil is passed it will force
-	// the server to send a remote config back.
-	LastRemoteConfigHash []byte
-
-	LastConnectionSettingsHash []byte
-
-	// The hash of the last locally-saved server-provided packages. If nil is passed
-	// it will force the server to send packages list back.
-	LastServerProvidedAllPackagesHash []byte
-}
-
 type OpAMPClient interface {
 
 	// Start the client and begin attempts to connect to the server. Once connection
@@ -67,7 +30,7 @@ type OpAMPClient interface {
 	//
 	// Start should be called only once. It should not be called concurrently with
 	// any other OpAMPClient methods.
-	Start(ctx context.Context, settings StartSettings) error
+	Start(ctx context.Context, settings types.StartSettings) error
 
 	// Stop the client. May be called only after Start() returns successfully.
 	// May be called only once.
diff --git a/client/httpclient.go b/client/httpclient.go
index b339970..d208d04 100644
--- a/client/httpclient.go
+++ b/client/httpclient.go
@@ -2,8 +2,6 @@ package client
 
 import (
 	"context"
-	"net/url"
-	"sync"
 
 	"github.com/open-telemetry/opamp-go/client/internal"
 	"github.com/open-telemetry/opamp-go/client/types"
@@ -14,159 +12,67 @@ import (
 // httpClient is an OpAMP Client implementation for plain HTTP transport.
 // See specification: https://github.com/open-telemetry/opamp-spec/blob/main/specification.md#plain-http-transport
 type httpClient struct {
-	logger   types.Logger
-	settings StartSettings
+	common internal.ClientCommon
 
-	// OpAMP Server URL.
-	url *url.URL
+	opAMPServerURL string
 
-	// True if Start() is successful.
-	isStarted bool
-
-	// Cancellation func for background go routines.
-	runCancel context.CancelFunc
-
-	// True when stopping is in progress.
-	isStoppingFlag  bool
-	isStoppingMutex sync.RWMutex
-
-	// Indicates that the Client is fully stopped.
-	stoppedSignal chan struct{}
-
-	// The looper performs HTTP request/response loop.
-	looper *internal.HTTPLooper
-
-	// Client state storage. This is needed if the Server asks to report the state.
-	clientSyncedState internal.ClientSyncedState
+	// The sender performs HTTP request/response loop.
+	sender *internal.HTTPSender
 }
 
-var _ OpAMPClient = (*httpClient)(nil)
-
 func NewHTTP(logger types.Logger) *httpClient {
 	if logger == nil {
 		logger = &sharedinternal.NopLogger{}
 	}
 
+	sender := internal.NewHTTPSender(logger)
 	w := &httpClient{
-		logger:        logger,
-		stoppedSignal: make(chan struct{}, 1),
+		common: internal.NewClientCommon(logger, sender),
+		sender: sender,
 	}
-	w.looper = internal.NewHTTPLooper(logger, &w.clientSyncedState)
-
 	return w
 }
 
-func (w *httpClient) Start(ctx context.Context, settings StartSettings) error {
-	if w.isStarted {
-		return errAlreadyStarted
-	}
-
-	if err := w.clientSyncedState.SetAgentDescription(settings.AgentDescription); err != nil {
+func (c *httpClient) Start(ctx context.Context, settings types.StartSettings) error {
+	if err := c.common.PrepareStart(ctx, settings); err != nil {
 		return err
 	}
 
-	w.settings = settings
-	if w.settings.Callbacks == nil {
-		// Make sure it is always safe to call Callbacks.
-		w.settings.Callbacks = types.CallbacksStruct{}
-	}
-
-	if err := w.looper.SetInstanceUid(w.settings.InstanceUid); err != nil {
-		return err
-	}
+	c.opAMPServerURL = settings.OpAMPServerURL
 
 	// Prepare server connection settings.
-	var err error
-	w.url, err = url.Parse(w.settings.OpAMPServerURL)
-	if err != nil {
-		return err
-	}
-
-	if w.settings.TLSConfig != nil {
-		w.url.Scheme = "https"
-	}
-
-	if w.settings.AuthorizationHeader != "" {
-		w.looper.SetRequestHeader("Authorization", w.settings.AuthorizationHeader)
+	if settings.AuthorizationHeader != "" {
+		c.sender.SetRequestHeader("Authorization", settings.AuthorizationHeader)
 	}
 
 	// Prepare the first message to send.
-	err = internal.PrepareFirstMessage(ctx, w.looper.NextMessage(), &w.clientSyncedState, w.settings.Callbacks)
+	err := c.common.PrepareFirstMessage(ctx)
 	if err != nil {
 		return err
 	}
 
-	w.looper.ScheduleSend()
-
-	w.startConnectAndRun()
-
-	w.isStarted = true
-
-	return nil
-}
-
-func (w *httpClient) Stop(ctx context.Context) error {
-	if !w.isStarted {
-		return errCannotStopNotStarted
-	}
-
-	w.isStoppingMutex.Lock()
-	cancelFunc := w.runCancel
-	w.isStoppingFlag = true
-	w.isStoppingMutex.Unlock()
+	c.sender.ScheduleSend()
 
-	cancelFunc()
+	c.common.StartConnectAndRun(c.runUntilStopped)
 
-	// Wait until stopping is finished.
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-w.stoppedSignal:
-	}
 	return nil
 }
 
-func (w *httpClient) SetAgentDescription(descr *protobufs.AgentDescription) error {
-	return internal.SetAgentDescription(w.looper, &w.clientSyncedState, descr)
+func (c *httpClient) Stop(ctx context.Context) error {
+	return c.common.Stop(ctx)
 }
 
-func (w *httpClient) UpdateEffectiveConfig(ctx context.Context) error {
-	return internal.UpdateEffectiveConfig(ctx, w.looper, w.settings.Callbacks)
+func (c *httpClient) SetAgentDescription(descr *protobufs.AgentDescription) error {
+	return c.common.SetAgentDescription(descr)
 }
 
-func (w *httpClient) isStopping() bool {
-	w.isStoppingMutex.RLock()
-	defer w.isStoppingMutex.RUnlock()
-	return w.isStoppingFlag
+func (c *httpClient) UpdateEffectiveConfig(ctx context.Context) error {
+	return c.common.UpdateEffectiveConfig(ctx)
 }
 
-func (w *httpClient) startConnectAndRun() {
-	// Create a cancellable context.
-	runCtx, runCancel := context.WithCancel(context.Background())
-
-	w.isStoppingMutex.Lock()
-	defer w.isStoppingMutex.Unlock()
-
-	if w.isStoppingFlag {
-		// Stop() was called. Don't connect.
-		runCancel()
-		return
-	}
-	w.runCancel = runCancel
-
-	go w.runUntilStopped(runCtx)
-}
-
-func (w *httpClient) runUntilStopped(ctx context.Context) {
-
-	defer func() {
-		// We only return from runUntilStopped when we are instructed to stop.
-		// When returning signal that we stopped.
-		w.stoppedSignal <- struct{}{}
-	}()
-
-	// Start the HTTP looper. This will make request/responses with retries for
+func (c *httpClient) runUntilStopped(ctx context.Context) {
+	// Start the HTTP sender. This will make request/responses with retries for
 	// failures and will wait with configured polling interval if there is nothing
 	// to send.
-	w.looper.Run(ctx, w.settings.OpAMPServerURL, w.settings.Callbacks)
+	c.sender.Run(ctx, c.opAMPServerURL, c.common.Callbacks, &c.common.ClientSyncedState)
 }
diff --git a/client/internal/clientcommon.go b/client/internal/clientcommon.go
index 7581e6c..21074a4 100644
--- a/client/internal/clientcommon.go
+++ b/client/internal/clientcommon.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -11,25 +12,120 @@ import (
 
 var (
 	errAgentDescriptionMissing = errors.New("AgentDescription is not set")
+	errAlreadyStarted          = errors.New("already started")
+	errCannotStopNotStarted    = errors.New("cannot stop because not started")
 )
 
-// This file contains the OpAMP logic that is common between WebSocket and
+// ClientCommon contains the OpAMP logic that is common between WebSocket and
 // plain HTTP transports.
-//
-// TODO: introduce a CommonClient struct that encapsulates this logic and
-// embed it in the HTTPClient and WSClient to avoid passing so many parameters
-// to these functions. There is more common OpAMP logic that we refactor and
-// move here.
+type ClientCommon struct {
+	Logger    types.Logger
+	Callbacks types.Callbacks
+
+	// Client state storage. This is needed if the Server asks to report the state.
+	ClientSyncedState ClientSyncedState
+
+	// The transport-specific sender.
+	sender Sender
+
+	// True if Start() is successful.
+	isStarted bool
+
+	// Cancellation func for background go routines.
+	runCancel context.CancelFunc
+
+	// True when stopping is in progress.
+	isStoppingFlag  bool
+	isStoppingMutex sync.RWMutex
+
+	// Indicates that the Client is fully stopped.
+	stoppedSignal chan struct{}
+}
+
+func NewClientCommon(logger types.Logger, sender Sender) ClientCommon {
+	return ClientCommon{Logger: logger, sender: sender, stoppedSignal: make(chan struct{}, 1)}
+}
+
+func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSettings) error {
+	if c.isStarted {
+		return errAlreadyStarted
+	}
+
+	if err := c.ClientSyncedState.SetAgentDescription(settings.AgentDescription); err != nil {
+		return err
+	}
+
+	c.Callbacks = settings.Callbacks
+	if c.Callbacks == nil {
+		// Make sure it is always safe to call Callbacks.
+		c.Callbacks = types.CallbacksStruct{}
+	}
+
+	if err := c.sender.SetInstanceUid(settings.InstanceUid); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *ClientCommon) Stop(ctx context.Context) error {
+	if !c.isStarted {
+		return errCannotStopNotStarted
+	}
+
+	c.isStoppingMutex.Lock()
+	cancelFunc := c.runCancel
+	c.isStoppingFlag = true
+	c.isStoppingMutex.Unlock()
+
+	cancelFunc()
+
+	// Wait until stopping is finished.
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-c.stoppedSignal:
+	}
+	return nil
+}
+
+func (c *ClientCommon) IsStopping() bool {
+	c.isStoppingMutex.RLock()
+	defer c.isStoppingMutex.RUnlock()
+	return c.isStoppingFlag
+}
+
+func (c *ClientCommon) StartConnectAndRun(runner func(ctx context.Context)) {
+	// Create a cancellable context.
+	runCtx, runCancel := context.WithCancel(context.Background())
+
+	c.isStoppingMutex.Lock()
+	defer c.isStoppingMutex.Unlock()
+
+	if c.isStoppingFlag {
+		// Stop() was called. Don't connect.
+		runCancel()
+		return
+	}
+	c.runCancel = runCancel
+
+	go func() {
+		defer func() {
+			// We only return from runner() when we are instructed to stop.
+			// When returning signal that we stopped.
+			c.stoppedSignal <- struct{}{}
+		}()
+
+		runner(runCtx)
+	}()
+
+	c.isStarted = true
+}
 
 // PrepareFirstMessage prepares the initial state of NextMessage struct that client
 // sends when it first establishes a connection with the server.
-func PrepareFirstMessage(
-	ctx context.Context,
-	nextMessage *NextMessage,
-	clientSyncedState *ClientSyncedState,
-	callbacks types.Callbacks,
-) error {
-	cfg, err := callbacks.GetEffectiveConfig(ctx)
+func (c *ClientCommon) PrepareFirstMessage(ctx context.Context) error {
+	cfg, err := c.Callbacks.GetEffectiveConfig(ctx)
 	if err != nil {
 		return err
 	}
@@ -37,12 +133,12 @@ func PrepareFirstMessage(
 		return errors.New("EffectiveConfig hash is empty")
 	}
 
-	nextMessage.Update(
+	c.sender.NextMessage().Update(
 		func(msg *protobufs.AgentToServer) {
 			if msg.StatusReport == nil {
 				msg.StatusReport = &protobufs.StatusReport{}
 			}
-			msg.StatusReport.AgentDescription = clientSyncedState.AgentDescription()
+			msg.StatusReport.AgentDescription = c.ClientSyncedState.AgentDescription()
 
 			msg.StatusReport.EffectiveConfig = cfg
 
@@ -60,23 +156,23 @@ func PrepareFirstMessage(
 // SetAgentDescription sends a status update to the Server with the new AgentDescription
 // and remembers the AgentDescription in the client state so that it can be sent
 // to the Server when the Server asks for it.
-func SetAgentDescription(sender Sender, clientSyncedState *ClientSyncedState, descr *protobufs.AgentDescription) error {
+func (c *ClientCommon) SetAgentDescription(descr *protobufs.AgentDescription) error {
 	// store the agent description to send on reconnect
-	if err := clientSyncedState.SetAgentDescription(descr); err != nil {
+	if err := c.ClientSyncedState.SetAgentDescription(descr); err != nil {
 		return err
 	}
-	sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
+	c.sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
 		statusReport.AgentDescription = descr
 	})
-	sender.ScheduleSend()
+	c.sender.ScheduleSend()
 	return nil
 }
 
 // UpdateEffectiveConfig fetches the current local effective config using
 // GetEffectiveConfig callback and sends it to the Server using provided Sender.
-func UpdateEffectiveConfig(ctx context.Context, sender Sender, callbacks types.Callbacks) error {
+func (c *ClientCommon) UpdateEffectiveConfig(ctx context.Context) error {
 	// Fetch the locally stored config.
-	cfg, err := callbacks.GetEffectiveConfig(ctx)
+	cfg, err := c.Callbacks.GetEffectiveConfig(ctx)
 	if err != nil {
 		return fmt.Errorf("GetEffectiveConfig failed: %w", err)
 	}
@@ -84,10 +180,10 @@ func UpdateEffectiveConfig(ctx context.Context, sender Sender, callbacks types.C
 		return errors.New("hash field must be set, use CalcHashEffectiveConfig")
 	}
 	// Send it to the server.
-	sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
+	c.sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
 		statusReport.EffectiveConfig = cfg
 	})
-	sender.ScheduleSend()
+	c.sender.ScheduleSend()
 
 	// Note that we do not store the EffectiveConfig anywhere else. It will be deleted
 	// from NextMessage when the message is sent. This avoids storing EffectiveConfig
diff --git a/client/internal/httplooper.go b/client/internal/httpsender.go
similarity index 71%
rename from client/internal/httplooper.go
rename to client/internal/httpsender.go
index de3cbbd..18a6587 100644
--- a/client/internal/httplooper.go
+++ b/client/internal/httpsender.go
@@ -21,13 +21,14 @@ import (
 const OpAMPPlainHTTPMethod = "POST"
 const defaultPollingIntervalMs = 30 * 1000 // default interval is 30 seconds.
 
-// HTTPLooper allows scheduling messages to send. Once run, it will loop through
+// HTTPSender allows scheduling messages to send. Once run, it will loop through
 // a request/response cycle for each message to send and will process all received
 // responses using a receivedProcessor. If there are no pending messages to send
-// the HTTPLooper will wait for the configured polling interval.
-type HTTPLooper struct {
+// the HTTPSender will wait for the configured polling interval.
+type HTTPSender struct {
+	SenderCommon
+
 	url               string
-	instanceUid       atomic.Value
 	logger            types.Logger
 	client            *http.Client
 	callbacks         types.Callbacks
@@ -36,27 +37,17 @@ type HTTPLooper struct {
 	// Headers to send with all requests.
 	requestHeader http.Header
 
-	// Indicates that there is a pending message to send.
-	hasPendingMessage chan struct{}
-
-	// The next message to send.
-	nextMessage NextMessage
-
 	// Processor to handle received messages.
 	receiveProcessor receivedProcessor
-
-	// Client state storage. This is needed if the Server asks to report the state.
-	clientSyncedState *ClientSyncedState
 }
 
-func NewHTTPLooper(logger types.Logger, clientSyncedState *ClientSyncedState) *HTTPLooper {
-	h := &HTTPLooper{
+func NewHTTPSender(logger types.Logger) *HTTPSender {
+	h := &HTTPSender{
+		SenderCommon:      NewSenderCommon(),
 		logger:            logger,
 		client:            http.DefaultClient,
-		hasPendingMessage: make(chan struct{}, 1),
 		requestHeader:     http.Header{},
 		pollingIntervalMs: defaultPollingIntervalMs,
-		clientSyncedState: clientSyncedState,
 	}
 	h.requestHeader.Set(headerContentType, contentTypeProtobuf)
 	return h
@@ -68,10 +59,10 @@ func NewHTTPLooper(logger types.Logger, clientSyncedState *ClientSyncedState) *H
 // Should not be called concurrently with itself. Can be called concurrently with
 // modifying NextMessage().
 // Run continues until ctx is cancelled.
-func (h *HTTPLooper) Run(ctx context.Context, url string, callbacks types.Callbacks) {
+func (h *HTTPSender) Run(ctx context.Context, url string, callbacks types.Callbacks, clientSyncedState *ClientSyncedState) {
 	h.url = url
 	h.callbacks = callbacks
-	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, h.clientSyncedState)
+	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState)
 
 	for {
 		pollingTimer := time.NewTimer(time.Millisecond * time.Duration(atomic.LoadInt64(&h.pollingIntervalMs)))
@@ -95,42 +86,13 @@ func (h *HTTPLooper) Run(ctx context.Context, url string, callbacks types.Callba
 	}
 }
 
-// ScheduleSend signals to HTTPLooper that the message in NextMessage struct
-// is now ready to be sent. If there is no pending message (e.g. the NextMessage was
-// already sent and "pending" flag is reset) then no message will be sent.
-func (h *HTTPLooper) ScheduleSend() {
-	// Set pending flag. Don't block on writing to channel.
-	select {
-	case h.hasPendingMessage <- struct{}{}:
-	default:
-		break
-	}
-}
-
-// SetInstanceUid sets a new instanceUid to be used for all subsequent messages to be sent.
-// Can be called concurrently, normally is called when a message is received from the
-// Server that instructs us to change our instance UID.
-func (h *HTTPLooper) SetInstanceUid(instanceUid string) error {
-	if instanceUid == "" {
-		return fmt.Errorf("cannot set instance uid to empty value")
-	}
-	h.instanceUid.Store(instanceUid)
-	return nil
-}
-
 // SetRequestHeader sets an additional HTTP header to send with all future requests.
 // Should not be called concurrently with any other method.
-func (h *HTTPLooper) SetRequestHeader(key, value string) {
+func (h *HTTPSender) SetRequestHeader(key, value string) {
 	h.requestHeader.Set(key, value)
 }
 
-// NextMessage gives access to the next message that will be sent by this looper.
-// Can be called concurrently with any other method.
-func (h *HTTPLooper) NextMessage() *NextMessage {
-	return &h.nextMessage
-}
-
-func (h *HTTPLooper) makeOneRequestRoundtrip(ctx context.Context) {
+func (h *HTTPSender) makeOneRequestRoundtrip(ctx context.Context) {
 	resp, err := h.sendRequestWithRetries(ctx)
 	if err != nil {
 		return
@@ -138,7 +100,7 @@ func (h *HTTPLooper) makeOneRequestRoundtrip(ctx context.Context) {
 	h.receiveResponse(ctx, resp)
 }
 
-func (h *HTTPLooper) sendRequestWithRetries(ctx context.Context) (*http.Response, error) {
+func (h *HTTPSender) sendRequestWithRetries(ctx context.Context) (*http.Response, error) {
 	req, err := h.prepareRequest(ctx)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
@@ -206,7 +168,7 @@ func recalculateInterval(interval time.Duration, resp *http.Response) time.Durat
 	return interval
 }
 
-func (h *HTTPLooper) prepareRequest(ctx context.Context) (*http.Request, error) {
+func (h *HTTPSender) prepareRequest(ctx context.Context) (*http.Request, error) {
 	msgToSend := h.nextMessage.PopPending()
 
 	if msgToSend == nil || proto.Equal(msgToSend, &protobufs.AgentToServer{}) {
@@ -215,8 +177,6 @@ func (h *HTTPLooper) prepareRequest(ctx context.Context) (*http.Request, error)
 		return nil, nil
 	}
 
-	msgToSend.InstanceUid = h.instanceUid.Load().(string)
-
 	data, err := proto.Marshal(msgToSend)
 	if err != nil {
 		return nil, err
@@ -232,7 +192,7 @@ func (h *HTTPLooper) prepareRequest(ctx context.Context) (*http.Request, error)
 	return req, nil
 }
 
-func (h *HTTPLooper) receiveResponse(ctx context.Context, resp *http.Response) {
+func (h *HTTPSender) receiveResponse(ctx context.Context, resp *http.Response) {
 	msgBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
 		_ = resp.Body.Close()
@@ -252,6 +212,6 @@ func (h *HTTPLooper) receiveResponse(ctx context.Context, resp *http.Response) {
 
 // SetPollingInterval sets the interval between polling. Has effect starting from the
 // next polling cycle.
-func (h *HTTPLooper) SetPollingInterval(duration time.Duration) {
+func (h *HTTPSender) SetPollingInterval(duration time.Duration) {
 	atomic.StoreInt64(&h.pollingIntervalMs, duration.Milliseconds())
 }
diff --git a/client/internal/sender.go b/client/internal/sender.go
index 04e57be..f463542 100644
--- a/client/internal/sender.go
+++ b/client/internal/sender.go
@@ -1,5 +1,11 @@
 package internal
 
+import (
+	"errors"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
 // Sender is an interface of the sending portion of OpAMP protocol that stores
 // the NextMessage to be sent and can be ordered to send the message.
 type Sender interface {
@@ -13,6 +19,55 @@ type Sender interface {
 	// "pending" flag is reset) then no message will be sent.
 	ScheduleSend()
 
-	// SetInstanceUid sets a new instanceUid to be used when NextMessage is being sent.
+	// SetInstanceUid sets a new instanceUid to be used for all subsequent messages to be sent.
 	SetInstanceUid(instanceUid string) error
 }
+
+// SenderCommon is partial Sender implementation that is common WebSocket and plain
+// HTTP transports. This struct is intended to be embedded in the WebSocket and
+// HTTP Sender implementations.
+type SenderCommon struct {
+	// Indicates that there is a pending message to send.
+	hasPendingMessage chan struct{}
+
+	// The next message to send.
+	nextMessage NextMessage
+}
+
+func NewSenderCommon() SenderCommon {
+	return SenderCommon{hasPendingMessage: make(chan struct{}, 1)}
+}
+
+// ScheduleSend signals to HTTPSender that the message in NextMessage struct
+// is now ready to be sent. If there is no pending message (e.g. the NextMessage was
+// already sent and "pending" flag is reset) then no message will be sent.
+func (h *SenderCommon) ScheduleSend() {
+	// Set pending flag. Don't block on writing to channel.
+	select {
+	case h.hasPendingMessage <- struct{}{}:
+	default:
+		break
+	}
+}
+
+// NextMessage gives access to the next message that will be sent by this looper.
+// Can be called concurrently with any other method.
+func (h *SenderCommon) NextMessage() *NextMessage {
+	return &h.nextMessage
+}
+
+// SetInstanceUid sets a new instanceUid to be used for all subsequent messages to be sent.
+// Can be called concurrently, normally is called when a message is received from the
+// Server that instructs us to change our instance UID.
+func (h *SenderCommon) SetInstanceUid(instanceUid string) error {
+	if instanceUid == "" {
+		return errors.New("cannot set instance uid to empty value")
+	}
+
+	h.nextMessage.Update(
+		func(msg *protobufs.AgentToServer) {
+			msg.InstanceUid = instanceUid
+		})
+
+	return nil
+}
diff --git a/client/internal/wssender.go b/client/internal/wssender.go
index f2f79d8..0ea8a49 100644
--- a/client/internal/wssender.go
+++ b/client/internal/wssender.go
@@ -2,8 +2,6 @@ package internal
 
 import (
 	"context"
-	"fmt"
-	"sync/atomic"
 
 	"github.com/gorilla/websocket"
 	"google.golang.org/protobuf/proto"
@@ -14,33 +12,24 @@ import (
 
 // WSSender implements the WebSocket client's sending portion of OpAMP protocol.
 type WSSender struct {
-	instanceUid atomic.Value
-	conn        *websocket.Conn
-
+	SenderCommon
+	conn   *websocket.Conn
 	logger types.Logger
-
-	// Indicates that there are pending messages to send.
-	hasMessages chan struct{}
-
-	// The next message to send.
-	nextMessage NextMessage
-
 	// Indicates that the sender has fully stopped.
 	stopped chan struct{}
 }
 
 func NewSender(logger types.Logger) *WSSender {
 	return &WSSender{
-		logger:      logger,
-		hasMessages: make(chan struct{}, 1),
+		logger:       logger,
+		SenderCommon: NewSenderCommon(),
 	}
 }
 
 // Start the sender and send the first message that was set via NextMessage().Update()
 // earlier. To stop the WSSender cancel the ctx.
-func (s *WSSender) Start(ctx context.Context, instanceUid string, conn *websocket.Conn) error {
+func (s *WSSender) Start(ctx context.Context, conn *websocket.Conn) error {
 	s.conn = conn
-	s.instanceUid.Store(instanceUid)
 	err := s.sendNextMessage()
 
 	// Run the sender in the background.
@@ -50,42 +39,17 @@ func (s *WSSender) Start(ctx context.Context, instanceUid string, conn *websocke
 	return err
 }
 
-// SetInstanceUid sets a new instanceUid without closing and reopening the connection. It will be used
-// when next message is being sent.
-func (s *WSSender) SetInstanceUid(instanceUid string) error {
-	if instanceUid == "" {
-		return fmt.Errorf("cannot set instance uid to empty value")
-	}
-	s.instanceUid.Store(instanceUid)
-	return nil
-}
-
-func (s *WSSender) NextMessage() *NextMessage {
-	return &s.nextMessage
-}
-
 // WaitToStop blocks until the sender is stopped. To stop the sender cancel the context
 // that was passed to Start().
 func (s *WSSender) WaitToStop() {
 	<-s.stopped
 }
 
-// ScheduleSend signals to the sending goroutine to send the next message
-// if it is pending. If there is no pending message (e.g. the message was
-// already sent and "pending" flag is reset) then no message will be be sent.
-func (s *WSSender) ScheduleSend() {
-	select {
-	case s.hasMessages <- struct{}{}:
-	default:
-		break
-	}
-}
-
 func (s *WSSender) run(ctx context.Context) {
 out:
 	for {
 		select {
-		case <-s.hasMessages:
+		case <-s.hasPendingMessage:
 			s.sendNextMessage()
 
 		case <-ctx.Done():
@@ -100,8 +64,6 @@ func (s *WSSender) sendNextMessage() error {
 	msgToSend := s.nextMessage.PopPending()
 	if msgToSend != nil && !proto.Equal(msgToSend, &protobufs.AgentToServer{}) {
 		// There is a pending message and the message has some fields populated.
-		// Set the InstanceUid field and send it.
-		msgToSend.InstanceUid = s.instanceUid.Load().(string)
 		return s.sendMessage(msgToSend)
 	}
 	return nil
diff --git a/client/types/startsettings.go b/client/types/startsettings.go
new file mode 100644
index 0000000..da7b88e
--- /dev/null
+++ b/client/types/startsettings.go
@@ -0,0 +1,43 @@
+package types
+
+import (
+	"crypto/tls"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+type StartSettings struct {
+	// Connection parameters.
+
+	// Server URL. MUST be set.
+	OpAMPServerURL string
+
+	// Optional value of "Authorization" HTTP header in the HTTP request.
+	AuthorizationHeader string
+
+	// Optional TLS config for HTTP connection.
+	TLSConfig *tls.Config
+
+	// Agent information.
+	InstanceUid string
+
+	// AgentDescription MUST be set and MUST describe the Agent. See OpAMP spec
+	// for details.
+	AgentDescription *protobufs.AgentDescription
+
+	// Callbacks that the client will call after Start() returns nil.
+	Callbacks Callbacks
+
+	// Previously saved state. These will be reported to the server immediately
+	// after the connection is established.
+
+	// The hash of the last received remote config. If nil is passed it will force
+	// the server to send a remote config back.
+	LastRemoteConfigHash []byte
+
+	LastConnectionSettingsHash []byte
+
+	// The hash of the last locally-saved server-provided packages. If nil is passed
+	// it will force the server to send packages list back.
+	LastServerProvidedAllPackagesHash []byte
+}
diff --git a/client/wsclient.go b/client/wsclient.go
index a1a1c79..da39369 100644
--- a/client/wsclient.go
+++ b/client/wsclient.go
@@ -17,16 +17,10 @@ import (
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-var (
-	errAlreadyStarted       = errors.New("already started")
-	errCannotStopNotStarted = errors.New("cannot stop because not started")
-)
-
 // wsClient is an OpAMP Client implementation for WebSocket transport.
 // See specification: https://github.com/open-telemetry/opamp-spec/blob/main/specification.md#websocket-transport
 type wsClient struct {
-	logger   types.Logger
-	settings StartSettings
+	common internal.ClientCommon
 
 	// OpAMP Server URL.
 	url *url.URL
@@ -39,137 +33,85 @@ type wsClient struct {
 	conn      *websocket.Conn
 	connMutex sync.RWMutex
 
-	// True if Start() is successful.
-	isStarted bool
-
-	// Cancellation func background running go routines.
-	runCancel context.CancelFunc
-
-	// True when stopping is in progress.
-	isStoppingFlag  bool
-	isStoppingMutex sync.RWMutex
-
-	// Indicates that the Client is fully stopped.
-	stoppedSignal chan struct{}
-
 	// The sender is responsible for sending portion of the OpAMP protocol.
 	sender *internal.WSSender
-
-	// Client state storage. This is needed if the Server asks to report the state.
-	clientSyncedState internal.ClientSyncedState
 }
 
-var _ OpAMPClient = (*wsClient)(nil)
-
 func NewWebSocket(logger types.Logger) *wsClient {
 	if logger == nil {
 		logger = &sharedinternal.NopLogger{}
 	}
 
+	sender := internal.NewSender(logger)
 	w := &wsClient{
-		logger:        logger,
-		stoppedSignal: make(chan struct{}, 1),
-		sender:        internal.NewSender(logger),
+		common: internal.NewClientCommon(logger, sender),
+		sender: sender,
 	}
 	return w
 }
 
-func (w *wsClient) Start(_ context.Context, settings StartSettings) error {
-	if w.isStarted {
-		return errAlreadyStarted
-	}
-
-	if err := w.clientSyncedState.SetAgentDescription(settings.AgentDescription); err != nil {
+func (c *wsClient) Start(ctx context.Context, settings types.StartSettings) error {
+	if err := c.common.PrepareStart(ctx, settings); err != nil {
 		return err
 	}
 
-	w.settings = settings
-	if w.settings.Callbacks == nil {
-		// Make sure it is always safe to call Callbacks.
-		w.settings.Callbacks = types.CallbacksStruct{}
-	}
+	// Prepare connection settings.
+	c.dialer = *websocket.DefaultDialer
 
 	var err error
-
-	// Prepare server connection settings.
-	w.url, err = url.Parse(w.settings.OpAMPServerURL)
+	c.url, err = url.Parse(settings.OpAMPServerURL)
 	if err != nil {
 		return err
 	}
 
-	w.dialer = *websocket.DefaultDialer
-
-	if w.settings.TLSConfig != nil {
-		w.url.Scheme = "wss"
+	if settings.TLSConfig != nil {
+		c.url.Scheme = "wss"
 	}
-	w.dialer.TLSClientConfig = w.settings.TLSConfig
+	c.dialer.TLSClientConfig = settings.TLSConfig
 
-	if w.settings.AuthorizationHeader != "" {
-		w.requestHeader = http.Header{}
-		w.requestHeader["Authorization"] = []string{w.settings.AuthorizationHeader}
+	if settings.AuthorizationHeader != "" {
+		c.requestHeader = http.Header{}
+		c.requestHeader["Authorization"] = []string{settings.AuthorizationHeader}
 	}
 
-	w.startConnectAndRun()
-
-	w.isStarted = true
+	c.common.StartConnectAndRun(c.runUntilStopped)
 
 	return nil
 }
 
-func (w *wsClient) Stop(ctx context.Context) error {
-	if !w.isStarted {
-		return errCannotStopNotStarted
-	}
-
-	w.isStoppingMutex.Lock()
-	cancelFunc := w.runCancel
-	w.isStoppingFlag = true
-	w.isStoppingMutex.Unlock()
-
-	cancelFunc()
-
+func (c *wsClient) Stop(ctx context.Context) error {
 	// Close connection if any.
-	w.connMutex.RLock()
-	conn := w.conn
-	w.connMutex.RUnlock()
+	c.connMutex.RLock()
+	conn := c.conn
+	c.connMutex.RUnlock()
 
 	if conn != nil {
 		_ = conn.Close()
 	}
-	// If we are not connected by this point we may still get connected because
-	// we are racing with tryConnectOnce() func. However, after tryConnectOnce we will
-	// check the isStopping flag in receiverLoop() and will exit receiverLoop, so we will
-	// not deadlock.
-
-	// Wait until stopping is finished.
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-w.stoppedSignal:
-	}
-	return nil
+
+	return c.common.Stop(ctx)
 }
 
-func (w *wsClient) SetAgentDescription(descr *protobufs.AgentDescription) error {
-	return internal.SetAgentDescription(w.sender, &w.clientSyncedState, descr)
+func (c *wsClient) SetAgentDescription(descr *protobufs.AgentDescription) error {
+	return c.common.SetAgentDescription(descr)
 }
 
-func (w *wsClient) UpdateEffectiveConfig(ctx context.Context) error {
-	return internal.UpdateEffectiveConfig(ctx, w.sender, w.settings.Callbacks)
+func (c *wsClient) UpdateEffectiveConfig(ctx context.Context) error {
+	return c.common.UpdateEffectiveConfig(ctx)
 }
 
 // Try to connect once. Returns an error if connection fails and optional retryAfter
 // duration to indicate to the caller to retry after the specified time as instructed
 // by the server.
-func (w *wsClient) tryConnectOnce(ctx context.Context) (err error, retryAfter sharedinternal.OptionalDuration) {
+func (c *wsClient) tryConnectOnce(ctx context.Context) (err error, retryAfter sharedinternal.OptionalDuration) {
 	var resp *http.Response
-	conn, resp, err := w.dialer.DialContext(ctx, w.url.String(), w.requestHeader)
+	conn, resp, err := c.dialer.DialContext(ctx, c.url.String(), c.requestHeader)
 	if err != nil {
-		if w.settings.Callbacks != nil {
-			w.settings.Callbacks.OnConnectFailed(err)
+		if c.common.Callbacks != nil {
+			c.common.Callbacks.OnConnectFailed(err)
 		}
 		if resp != nil {
-			w.logger.Errorf("Server responded with status=%v", resp.Status)
+			c.common.Logger.Errorf("Server responded with status=%v", resp.Status)
 			duration := sharedinternal.ExtractRetryAfterHeader(resp)
 			return err, duration
 		}
@@ -177,11 +119,11 @@ func (w *wsClient) tryConnectOnce(ctx context.Context) (err error, retryAfter sh
 	}
 
 	// Successfully connected.
-	w.connMutex.Lock()
-	w.conn = conn
-	w.connMutex.Unlock()
-	if w.settings.Callbacks != nil {
-		w.settings.Callbacks.OnConnect()
+	c.connMutex.Lock()
+	c.conn = conn
+	c.connMutex.Unlock()
+	if c.common.Callbacks != nil {
+		c.common.Callbacks.OnConnect()
 	}
 
 	return nil, sharedinternal.OptionalDuration{Defined: false}
@@ -189,7 +131,7 @@ func (w *wsClient) tryConnectOnce(ctx context.Context) (err error, retryAfter sh
 
 // Continuously try until connected. Will return nil when successfully
 // connected. Will return error if it is cancelled via context.
-func (w *wsClient) ensureConnected(ctx context.Context) error {
+func (c *wsClient) ensureConnected(ctx context.Context) error {
 	infiniteBackoff := backoff.NewExponentialBackOff()
 
 	// Make ticker run forever.
@@ -204,12 +146,12 @@ func (w *wsClient) ensureConnected(ctx context.Context) error {
 		select {
 		case <-timer.C:
 			{
-				if err, retryAfter := w.tryConnectOnce(ctx); err != nil {
+				if err, retryAfter := c.tryConnectOnce(ctx); err != nil {
 					if errors.Is(err, context.Canceled) {
-						w.logger.Debugf("Client is stopped, will not try anymore.")
+						c.common.Logger.Debugf("Client is stopped, will not try anymore.")
 						return err
 					} else {
-						w.logger.Errorf("Connection failed (%v), will retry.", err)
+						c.common.Logger.Errorf("Connection failed (%v), will retry.", err)
 					}
 					// Retry again a bit later.
 
@@ -227,58 +169,35 @@ func (w *wsClient) ensureConnected(ctx context.Context) error {
 			}
 
 		case <-ctx.Done():
-			w.logger.Debugf("Client is stopped, will not try anymore.")
+			c.common.Logger.Debugf("Client is stopped, will not try anymore.")
 			timer.Stop()
 			return ctx.Err()
 		}
 	}
 }
 
-func (w *wsClient) isStopping() bool {
-	w.isStoppingMutex.RLock()
-	defer w.isStoppingMutex.RUnlock()
-	return w.isStoppingFlag
-}
-
-func (w *wsClient) startConnectAndRun() {
-	// Create a cancellable context.
-	runCtx, runCancel := context.WithCancel(context.Background())
-
-	w.isStoppingMutex.Lock()
-	defer w.isStoppingMutex.Unlock()
-
-	if w.isStoppingFlag {
-		// Stop() was called. Don't connect.
-		runCancel()
-		return
-	}
-	w.runCancel = runCancel
-
-	go w.runUntilStopped(runCtx)
-}
-
 // runOneCycle performs the following actions:
 //   1. connect (try until succeeds).
 //   2. send first status report.
 //   3. receive and process messages until error happens.
 // If it encounters an error it closes the connection and returns.
 // Will stop and return if Stop() is called (ctx is cancelled, isStopping is set).
-func (w *wsClient) runOneCycle(ctx context.Context) {
-	if err := w.ensureConnected(ctx); err != nil {
+func (c *wsClient) runOneCycle(ctx context.Context) {
+	if err := c.ensureConnected(ctx); err != nil {
 		// Can't connect, so can't move forward. This currently happens when we
 		// are being stopped.
 		return
 	}
 
-	if w.isStopping() {
-		_ = w.conn.Close()
+	if c.common.IsStopping() {
+		_ = c.conn.Close()
 		return
 	}
 
 	// Prepare the first status report.
-	err := internal.PrepareFirstMessage(ctx, w.sender.NextMessage(), &w.clientSyncedState, w.settings.Callbacks)
+	err := c.common.PrepareFirstMessage(ctx)
 	if err != nil {
-		w.logger.Errorf("cannot prepare the first message:%v", err)
+		c.common.Logger.Errorf("cannot prepare the first message:%v", err)
 		return
 	}
 
@@ -287,16 +206,16 @@ func (w *wsClient) runOneCycle(ctx context.Context) {
 
 	// Connected successfully. Start the sender. This will also send the first
 	// status report.
-	if err := w.sender.Start(procCtx, w.settings.InstanceUid, w.conn); err != nil {
-		w.logger.Errorf("Failed to send first status report: %v", err)
+	if err := c.sender.Start(procCtx, c.conn); err != nil {
+		c.common.Logger.Errorf("Failed to send first status report: %v", err)
 		// We could not send the report, the only thing we can do is start over.
-		_ = w.conn.Close()
+		_ = c.conn.Close()
 		procCancel()
 		return
 	}
 
 	// First status report sent. Now loop to receive and process messages.
-	r := internal.NewWSReceiver(w.logger, w.settings.Callbacks, w.conn, w.sender, &w.clientSyncedState)
+	r := internal.NewWSReceiver(c.common.Logger, c.common.Callbacks, c.conn, c.sender, &c.common.ClientSyncedState)
 	r.ReceiverLoop(ctx)
 
 	// Stop the background processors.
@@ -306,26 +225,19 @@ func (w *wsClient) runOneCycle(ctx context.Context) {
 	// read messages anymore. We need to start over.
 
 	// Close the connection to unblock the WSSender as well.
-	_ = w.conn.Close()
+	_ = c.conn.Close()
 
 	// Wait for WSSender to stop.
-	w.sender.WaitToStop()
+	c.sender.WaitToStop()
 }
 
-func (w *wsClient) runUntilStopped(ctx context.Context) {
-
-	defer func() {
-		// We only return from runUntilStopped when we are instructed to stop.
-		// When returning signal that we stopped.
-		w.stoppedSignal <- struct{}{}
-	}()
-
+func (c *wsClient) runUntilStopped(ctx context.Context) {
 	// Iterates until we detect that the client is stopping.
 	for {
-		if w.isStopping() {
+		if c.common.IsStopping() {
 			return
 		}
 
-		w.runOneCycle(ctx)
+		c.runOneCycle(ctx)
 	}
 }
diff --git a/internal/examples/agent/agent/agent.go b/internal/examples/agent/agent/agent.go
index 137872c..76a5a46 100644
--- a/internal/examples/agent/agent/agent.go
+++ b/internal/examples/agent/agent/agent.go
@@ -83,7 +83,7 @@ func NewAgent(logger types.Logger, agentType string, agentVersion string) *Agent
 func (agent *Agent) start() error {
 	agent.opampClient = client.NewWebSocket(agent.logger)
 
-	settings := client.StartSettings{
+	settings := types.StartSettings{
 		OpAMPServerURL:   "ws://127.0.0.1:4320/v1/opamp",
 		InstanceUid:      agent.instanceId.String(),
 		AgentDescription: agent.agentDescription,
@@ -180,7 +180,7 @@ func (agent *Agent) updateAgentIdentity(instanceId ulid.ULID) {
 
 func (agent *Agent) loadLocalConfig() {
 	var k = koanf.New(".")
-	k.Load(rawbytes.Provider([]byte(localConfig)), yaml.Parser())
+	_ = k.Load(rawbytes.Provider([]byte(localConfig)), yaml.Parser())
 
 	effectiveConfigBytes, err := k.Marshal(yaml.Parser())
 	if err != nil {
@@ -353,6 +353,6 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (conf
 func (agent *Agent) Shutdown() {
 	agent.logger.Debugf("Agent shutting down...")
 	if agent.opampClient != nil {
-		agent.opampClient.Stop(context.Background())
+		_ = agent.opampClient.Stop(context.Background())
 	}
 }


Test patch:
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
index 7dc67c2..3c2bb6a 100644
--- a/client/clientimpl_test.go
+++ b/client/clientimpl_test.go
@@ -64,7 +64,7 @@ func testClients(t *testing.T, f func(client OpAMPClient)) {
 
 func TestConnectInvalidURL(t *testing.T) {
 	testClients(t, func(client OpAMPClient) {
-		settings := StartSettings{
+		settings := types.StartSettings{
 			OpAMPServerURL: ":not a url",
 		}
 
@@ -77,7 +77,7 @@ func eventually(t *testing.T, f func() bool) {
 	assert.Eventually(t, f, 5*time.Second, 10*time.Millisecond)
 }
 
-func prepareClient(t *testing.T, settings *StartSettings, c OpAMPClient) {
+func prepareClient(t *testing.T, settings *types.StartSettings, c OpAMPClient) {
 	// Autogenerate instance id.
 	entropy := ulid.Monotonic(rand.New(rand.NewSource(99)), 0)
 	settings.InstanceUid = ulid.MustNew(ulid.Timestamp(time.Now()), entropy).String()
@@ -96,15 +96,15 @@ func prepareClient(t *testing.T, settings *StartSettings, c OpAMPClient) {
 	settings.AgentDescription = createAgentDescr()
 }
 
-func startClient(t *testing.T, settings StartSettings, client OpAMPClient) {
+func startClient(t *testing.T, settings types.StartSettings, client OpAMPClient) {
 	prepareClient(t, &settings, client)
 	err := client.Start(context.Background(), settings)
 	assert.NoError(t, err)
 }
 
 // Create start settings that point to a non-existing server.
-func createNoServerSettings() StartSettings {
-	return StartSettings{
+func createNoServerSettings() types.StartSettings {
+	return types.StartSettings{
 		OpAMPServerURL: "ws://" + testhelpers.GetAvailableLocalAddress(),
 	}
 }
@@ -177,7 +177,7 @@ func TestConnectWithServer(t *testing.T) {
 
 		// Start a client.
 		var connected int64
-		settings := StartSettings{
+		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
 				OnConnectFunc: func() {
 					atomic.StoreInt64(&connected, 1)
@@ -215,7 +215,7 @@ func TestConnectWithServer503(t *testing.T) {
 		// Start a client.
 		var clientConnected int64
 		var connectErr atomic.Value
-		settings := StartSettings{
+		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
 				OnConnectFunc: func() {
 					atomic.StoreInt64(&clientConnected, 1)
@@ -253,7 +253,7 @@ func TestConnectWithHeader(t *testing.T) {
 		}
 
 		// Start a client.
-		settings := StartSettings{
+		settings := types.StartSettings{
 			OpAMPServerURL:      "ws://" + srv.Endpoint,
 			AuthorizationHeader: "Bearer 12345678",
 		}
@@ -289,7 +289,7 @@ func TestFirstStatusReport(t *testing.T) {
 
 		// Start a client.
 		var connected, remoteConfigReceived int64
-		settings := StartSettings{
+		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
 				OnConnectFunc: func() {
 					atomic.AddInt64(&connected, 1)
@@ -345,7 +345,7 @@ func TestIncludesDetailsOnReconnect(t *testing.T) {
 	}
 
 	var connected int64
-	settings := StartSettings{
+	settings := types.StartSettings{
 		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.AddInt64(&connected, 1)
@@ -406,7 +406,7 @@ func TestSetEffectiveConfig(t *testing.T) {
 
 		// Start a client.
 		sendConfig := createEffectiveConfig()
-		settings := StartSettings{
+		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
 				GetEffectiveConfigFunc: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
 					return sendConfig, nil
@@ -466,7 +466,7 @@ func TestSetAgentDescription(t *testing.T) {
 		}
 
 		// Start a client.
-		settings := StartSettings{
+		settings := types.StartSettings{
 			OpAMPServerURL: "ws://" + srv.Endpoint,
 		}
 		prepareClient(t, &settings, client)
@@ -538,7 +538,7 @@ func TestAgentIdentification(t *testing.T) {
 		}
 
 		// Start a client.
-		settings := StartSettings{}
+		settings := types.StartSettings{}
 		settings.OpAMPServerURL = "ws://" + srv.Endpoint
 		settings.AgentDescription = createAgentDescr()
 		settings.Callbacks = types.CallbacksStruct{
@@ -628,7 +628,7 @@ func TestConnectionSettings(t *testing.T) {
 		var gotOtherSettings int64
 
 		// Start a client.
-		settings := StartSettings{
+		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
 				OnOpampConnectionSettingsFunc: func(
 					ctx context.Context, settings *protobufs.ConnectionSettings,
@@ -692,7 +692,7 @@ func TestReportAgentDescription(t *testing.T) {
 		srv.EnableExpectMode()
 
 		// Start a client.
-		settings := StartSettings{
+		settings := types.StartSettings{
 			OpAMPServerURL: "ws://" + srv.Endpoint,
 		}
 		prepareClient(t, &settings, client)
@@ -758,7 +758,7 @@ func TestReportEffectiveConfig(t *testing.T) {
 		clientEffectiveConfig := createEffectiveConfig()
 
 		// Start a client.
-		settings := StartSettings{
+		settings := types.StartSettings{
 			OpAMPServerURL: "ws://" + srv.Endpoint,
 			Callbacks: types.CallbacksStruct{
 				GetEffectiveConfigFunc: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
diff --git a/client/httpclient_test.go b/client/httpclient_test.go
index d0093b4..0b0bcce 100644
--- a/client/httpclient_test.go
+++ b/client/httpclient_test.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/open-telemetry/opamp-go/client/internal"
+	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
 	"github.com/stretchr/testify/assert"
 )
@@ -23,13 +24,13 @@ func TestHTTPPolling(t *testing.T) {
 	}
 
 	// Start a client.
-	settings := StartSettings{}
+	settings := types.StartSettings{}
 	settings.OpAMPServerURL = "http://" + srv.Endpoint
 	client := NewHTTP(nil)
 	prepareClient(t, &settings, client)
 
 	// Shorten the polling interval to speed up the test.
-	client.looper.SetPollingInterval(time.Millisecond * 10)
+	client.sender.SetPollingInterval(time.Millisecond * 10)
 
 	assert.NoError(t, client.Start(context.Background(), settings))
 
diff --git a/client/wsclient_test.go b/client/wsclient_test.go
index 0f69732..3752297 100644
--- a/client/wsclient_test.go
+++ b/client/wsclient_test.go
@@ -22,7 +22,7 @@ func TestDisconnectWSByServer(t *testing.T) {
 	// Start an OpAMP/WebSocket client.
 	var connected int64
 	var connectErr atomic.Value
-	settings := StartSettings{
+	settings := types.StartSettings{
 		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.StoreInt64(&connected, 1)
@@ -42,7 +42,7 @@ func TestDisconnectWSByServer(t *testing.T) {
 
 	// Close the server and forcefully disconnect.
 	srv.Close()
-	conn.Load().(*websocket.Conn).Close()
+	_ = conn.Load().(*websocket.Conn).Close()
 
 	// The client must retry and must fail now.
 	eventually(t, func() bool { return connectErr.Load() != nil })