Repository: open-telemetry/opamp-go
Pull request #20 (2021-11-23 15:43:24)

Implement OpAMP Client Basics
Resolves: https://github.com/open-telemetry/opamp-go/issues/17
Resolves: https://github.com/open-telemetry/opamp-go/issues/10
Resolves: https://github.com/open-telemetry/opamp-go/issues/8
Resolves: https://github.com/open-telemetry/opamp-go/issues/11

Implemented the following Client features:
- Connection establishment, retries, backoff.
- Status reporting.
- Remote configuration receiving.
- Connection settings receiving.

tigrannajaryan: @djaglowski @MovieStoreGuy @pmm-sumo thanks a lot for reviews. Please take another look.
tigrannajaryan: @MovieStoreGuy Please let me know if any of the outstanding comments you have are blockers, if not I will merge this PR and we can address them later.

Patch:
diff --git a/client/client.go b/client/client.go
index 288383d..6926482 100644
--- a/client/client.go
+++ b/client/client.go
@@ -65,13 +65,19 @@ type OpAMPClient interface {
 	//  - OnConnectFailed
 	//  - OnError
 	//  - OnRemoteConfig
+	//
+	// Start should be called only once. It should not be called concurrently with
+	// any other OpAMPClient methods.
 	Start(settings StartSettings) error
 
-	// Stop the client. After this call returns successfully it is guaranteed
-	// that no callbacks will be called. Stop() will cancel context of any in-fly
+	// Stop the client. May be called only after Start() returns successfully.
+	// May be called only once.
+	// After this call returns successfully it is guaranteed that no
+	// callbacks will be called. Stop() will cancel context of any in-fly
 	// callbacks, but will wait until such in-fly callbacks are returned before
 	// Stop returns, so make sure the callbacks don't block infinitely and react
 	// promptly to context cancellations.
+	// Once stopped OpAMPClient cannot be started again.
 	Stop(ctx context.Context) error
 
 	// SetAgentAttributes sets attributes of the Agent. The attributes will be included
diff --git a/client/clientimpl.go b/client/clientimpl.go
new file mode 100644
index 0000000..28cef3f
--- /dev/null
+++ b/client/clientimpl.go
@@ -0,0 +1,352 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/gorilla/websocket"
+
+	"github.com/open-telemetry/opamp-go/client/internal"
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+var (
+	errAlreadyStarted       = errors.New("already started")
+	errCannotStopNotStarted = errors.New("cannot stop because not started")
+)
+
+// client is a Client implementation.
+type client struct {
+	logger   types.Logger
+	settings StartSettings
+
+	// OpAMP Server URL.
+	url *url.URL
+
+	// HTTP request headers to use when connecting to OpAMP Server.
+	requestHeader http.Header
+
+	// Websocket dialer and connection.
+	dialer    websocket.Dialer
+	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.Sender
+}
+
+var _ OpAMPClient = (*client)(nil)
+
+const retryAfterHTTPHeader = "Retry-After"
+
+func New(logger types.Logger) *client {
+	if logger == nil {
+		logger = &nopLogger{}
+	}
+
+	w := &client{
+		logger:        logger,
+		stoppedSignal: make(chan struct{}, 1),
+		sender:        internal.NewSender(logger),
+	}
+	return w
+}
+
+func (w *client) Start(settings StartSettings) error {
+	if w.isStarted {
+		return errAlreadyStarted
+	}
+
+	w.settings = settings
+
+	var err error
+
+	// Prepare server connection settings.
+	w.url, err = url.Parse(w.settings.OpAMPServerURL)
+	if err != nil {
+		return err
+	}
+
+	w.dialer = *websocket.DefaultDialer
+
+	if w.settings.TLSConfig != nil {
+		w.url.Scheme = "wss"
+	}
+	w.dialer.TLSClientConfig = w.settings.TLSConfig
+
+	if w.settings.AuthorizationHeader != "" {
+		w.requestHeader = http.Header{}
+		w.requestHeader["Authorization"] = []string{w.settings.AuthorizationHeader}
+	}
+
+	// Prepare the first status report.
+	w.sender.UpdateStatus(
+		func(statusReport *protobufs.StatusReport) {
+			statusReport.AgentDescription = &protobufs.AgentDescription{
+				AgentType:    w.settings.AgentType,
+				AgentVersion: w.settings.AgentVersion,
+			}
+
+			statusReport.ServerProvidedAllAddonsHash = w.settings.LastServerProvidedAllAddonsHash
+		},
+	)
+
+	w.startConnectAndRun()
+
+	w.isStarted = true
+
+	return nil
+}
+
+func (w *client) Stop(ctx context.Context) error {
+	if !w.isStarted {
+		return errCannotStopNotStarted
+	}
+
+	w.isStoppingMutex.Lock()
+	cancelFunc := w.runCancel
+	w.isStoppingFlag = true
+	w.isStoppingMutex.Unlock()
+
+	cancelFunc()
+
+	// Close connection if any.
+	w.connMutex.RLock()
+	conn := w.conn
+	w.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
+}
+
+func (w *client) SetAgentAttributes(attrs map[string]protobufs.AnyValue) error {
+	panic("implement me")
+}
+
+func (w *client) SetEffectiveConfig(config *protobufs.EffectiveConfig) error {
+	w.sender.UpdateStatus(func(statusReport *protobufs.StatusReport) {
+		statusReport.EffectiveConfig = config
+	})
+	w.sender.ScheduleSend()
+	return nil
+}
+
+type optionalDuration struct {
+	duration time.Duration
+	// true if duration field is defined.
+	defined bool
+}
+
+// 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 *client) tryConnectOnce(ctx context.Context) (err error, retryAfter optionalDuration) {
+	var resp *http.Response
+	conn, resp, err := w.dialer.DialContext(ctx, w.url.String(), w.requestHeader)
+	if err != nil {
+		if w.settings.Callbacks != nil {
+			w.settings.Callbacks.OnConnectFailed(err)
+		}
+		if resp != nil {
+			w.logger.Errorf("Server responded with status=%v", resp.Status)
+			duration := extractRetryAfterHeader(resp)
+			return err, duration
+		}
+		return err, optionalDuration{defined: false}
+	}
+
+	// Successfully connected.
+	w.connMutex.Lock()
+	w.conn = conn
+	w.connMutex.Unlock()
+	if w.settings.Callbacks != nil {
+		w.settings.Callbacks.OnConnect()
+	}
+
+	return nil, optionalDuration{defined: false}
+}
+
+// Extract Retry-After response header if the status is 503 or 429. Returns
+// 0 duration if the header is not found or the status is different.
+func extractRetryAfterHeader(resp *http.Response) optionalDuration {
+	if resp.StatusCode == http.StatusServiceUnavailable ||
+		resp.StatusCode == http.StatusTooManyRequests {
+		retryAfter := strings.TrimSpace(resp.Header.Get(retryAfterHTTPHeader))
+		if retryAfter != "" {
+			retryIntervalSec, err := strconv.Atoi(retryAfter)
+			if err == nil {
+				retryInterval := time.Duration(retryIntervalSec) * time.Second
+				return optionalDuration{defined: true, duration: retryInterval}
+			}
+		}
+	}
+	return optionalDuration{defined: false}
+}
+
+// Continuously try until connected. Will return nil when successfully
+// connected. Will return error if it is cancelled via context.
+func (w *client) ensureConnected(ctx context.Context) error {
+	infiniteBackoff := backoff.NewExponentialBackOff()
+
+	// Make ticker run forever.
+	infiniteBackoff.MaxElapsedTime = 0
+
+	interval := time.Duration(0)
+
+	for {
+		timer := time.NewTimer(interval)
+		interval = infiniteBackoff.NextBackOff()
+
+		select {
+		case <-timer.C:
+			{
+				if err, retryAfter := w.tryConnectOnce(ctx); err != nil {
+					if errors.Is(err, context.Canceled) {
+						w.logger.Debugf("Client is stopped, will not try anymore.")
+						return err
+					} else {
+						w.logger.Errorf("Connection failed (%v), will retry.", err)
+					}
+					// Retry again a bit later.
+
+					if retryAfter.defined && retryAfter.duration > interval {
+						// If the server suggested connecting later than our interval
+						// then honour server's request, otherwise wait at least
+						// as much as we calculated.
+						interval = retryAfter.duration
+					}
+
+					continue
+				}
+				// Connected successfully.
+				return nil
+			}
+
+		case <-ctx.Done():
+			w.logger.Debugf("Client is stopped, will not try anymore.")
+			timer.Stop()
+			return ctx.Err()
+		}
+	}
+}
+
+func (w *client) isStopping() bool {
+	w.isStoppingMutex.RLock()
+	defer w.isStoppingMutex.RUnlock()
+	return w.isStoppingFlag
+}
+
+func (w *client) 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.
+		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 *client) runOneCycle(ctx context.Context) {
+	if err := w.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()
+		return
+	}
+
+	// Create a cancellable context for background processors.
+	procCtx, procCancel := context.WithCancel(ctx)
+
+	// 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)
+		// We could not send the report, the only thing we can do is start over.
+		w.conn.Close()
+		return
+	}
+
+	// First status report sent. Now loop to receive and process messages.
+	r := internal.NewReceiver(w.logger, w.settings.Callbacks, w.conn, w.sender)
+	r.ReceiverLoop(ctx)
+
+	// Stop the background processors.
+	procCancel()
+
+	// If we exited receiverLoop it means there is a connection error, we cannot
+	// read messages anymore. We need to start over.
+
+	// Close the connection to unblock the Sender as well.
+	w.conn.Close()
+
+	// Wait for Sender to stop.
+	w.sender.WaitToStop()
+}
+
+func (w *client) 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{}{}
+	}()
+
+	// Iterates until we detect that the client is stopping.
+	for {
+		if w.isStopping() {
+			return
+		}
+
+		w.runOneCycle(ctx)
+	}
+}
diff --git a/client/internal/callbacks.go b/client/internal/callbacks.go
new file mode 100644
index 0000000..9aed5c8
--- /dev/null
+++ b/client/internal/callbacks.go
@@ -0,0 +1,125 @@
+package internal
+
+import (
+	"context"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+type CallbacksStruct struct {
+	OnConnectFunc       func()
+	OnConnectFailedFunc func(err error)
+	OnErrorFunc         func(err *protobufs.ServerErrorResponse)
+
+	OnRemoteConfigFunc func(
+		ctx context.Context,
+		config *protobufs.AgentRemoteConfig,
+	) (*protobufs.EffectiveConfig, error)
+
+	OnOpampConnectionSettingsFunc func(
+		ctx context.Context,
+		settings *protobufs.ConnectionSettings,
+	) error
+	OnOpampConnectionSettingsAcceptedFunc func(
+		settings *protobufs.ConnectionSettings,
+	)
+
+	OnOwnTelemetryConnectionSettingsFunc func(
+		ctx context.Context,
+		telemetryType types.OwnTelemetryType,
+		settings *protobufs.ConnectionSettings,
+	) error
+
+	OnOtherConnectionSettingsFunc func(
+		ctx context.Context,
+		name string,
+		settings *protobufs.ConnectionSettings,
+	) error
+
+	OnAddonsAvailableFunc       func(ctx context.Context, addons *protobufs.AddonsAvailable, syncer types.AddonSyncer) error
+	OnAgentPackageAvailableFunc func(addons *protobufs.AgentPackageAvailable, syncer types.AgentPackageSyncer) error
+}
+
+var _ types.Callbacks = (*CallbacksStruct)(nil)
+
+func (c CallbacksStruct) OnConnect() {
+	if c.OnConnectFunc != nil {
+		c.OnConnectFunc()
+	}
+}
+
+func (c CallbacksStruct) OnConnectFailed(err error) {
+	if c.OnConnectFailedFunc != nil {
+		c.OnConnectFailedFunc(err)
+	}
+}
+
+func (c CallbacksStruct) OnError(err *protobufs.ServerErrorResponse) {
+	if c.OnErrorFunc != nil {
+		c.OnErrorFunc(err)
+	}
+}
+
+func (c CallbacksStruct) OnRemoteConfig(
+	ctx context.Context, remoteConfig *protobufs.AgentRemoteConfig,
+) (*protobufs.EffectiveConfig, error) {
+	if c.OnRemoteConfigFunc != nil {
+		return c.OnRemoteConfigFunc(ctx, remoteConfig)
+	}
+	return nil, nil
+}
+
+func (c CallbacksStruct) OnOpampConnectionSettings(
+	ctx context.Context, settings *protobufs.ConnectionSettings,
+) error {
+	if c.OnOpampConnectionSettingsFunc != nil {
+		return c.OnOpampConnectionSettingsFunc(ctx, settings)
+	}
+	return nil
+}
+
+func (c CallbacksStruct) OnOpampConnectionSettingsAccepted(settings *protobufs.ConnectionSettings) {
+	if c.OnOpampConnectionSettingsAcceptedFunc != nil {
+		c.OnOpampConnectionSettingsAcceptedFunc(settings)
+	}
+}
+
+func (c CallbacksStruct) OnOwnTelemetryConnectionSettings(
+	ctx context.Context, telemetryType types.OwnTelemetryType,
+	settings *protobufs.ConnectionSettings,
+) error {
+	if c.OnOwnTelemetryConnectionSettingsFunc != nil {
+		return c.OnOwnTelemetryConnectionSettingsFunc(ctx, telemetryType, settings)
+	}
+	return nil
+}
+
+func (c CallbacksStruct) OnOtherConnectionSettings(
+	ctx context.Context, name string, settings *protobufs.ConnectionSettings,
+) error {
+	if c.OnOtherConnectionSettingsFunc != nil {
+		return c.OnOtherConnectionSettingsFunc(ctx, name, settings)
+	}
+	return nil
+}
+
+func (c CallbacksStruct) OnAddonsAvailable(
+	ctx context.Context,
+	addons *protobufs.AddonsAvailable,
+	syncer types.AddonSyncer,
+) error {
+	if c.OnAddonsAvailableFunc != nil {
+		return c.OnAddonsAvailableFunc(ctx, addons, syncer)
+	}
+	return nil
+}
+
+func (c CallbacksStruct) OnAgentPackageAvailable(
+	addons *protobufs.AgentPackageAvailable, syncer types.AgentPackageSyncer,
+) error {
+	if c.OnAgentPackageAvailableFunc != nil {
+		return c.OnAgentPackageAvailableFunc(addons, syncer)
+	}
+	return nil
+}
diff --git a/client/internal/mockserver.go b/client/internal/mockserver.go
new file mode 100644
index 0000000..f9e8066
--- /dev/null
+++ b/client/internal/mockserver.go
@@ -0,0 +1,92 @@
+package internal
+
+import (
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+	"github.com/open-telemetry/opamp-go/internal/testhelpers"
+)
+
+type MockServer struct {
+	Endpoint  string
+	OnRequest func(w http.ResponseWriter, r *http.Request)
+	OnConnect func(r *http.Request, conn *websocket.Conn)
+	OnMessage func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent
+	srv       *httptest.Server
+}
+
+var upgrader = websocket.Upgrader{}
+
+func StartMockServer(t *testing.T) *MockServer {
+	srv := &MockServer{}
+
+	m := http.NewServeMux()
+	m.HandleFunc(
+		"/", func(w http.ResponseWriter, r *http.Request) {
+			if srv.OnRequest != nil {
+				srv.OnRequest(w, r)
+				return
+			}
+
+			conn, err := upgrader.Upgrade(w, r, nil)
+			if err != nil {
+				return
+			}
+			if srv.OnConnect != nil {
+				srv.OnConnect(r, conn)
+			}
+			for {
+				var messageType int
+				var msgBytes []byte
+				if messageType, msgBytes, err = conn.ReadMessage(); err != nil {
+					return
+				}
+				assert.EqualValues(t, websocket.BinaryMessage, messageType)
+
+				var dest protobufs.AgentToServer
+				err := proto.Unmarshal(msgBytes, &dest)
+				if err != nil {
+					log.Fatal("cannot decode:", err)
+				}
+
+				if srv.OnMessage != nil {
+					response := srv.OnMessage(&dest)
+					if response != nil {
+						msgBytes, err := proto.Marshal(response)
+						if err != nil {
+							log.Fatal("cannot encode:", err)
+						}
+						err = conn.WriteMessage(websocket.BinaryMessage, msgBytes)
+						if err != nil {
+							log.Fatal("cannot send:", err)
+						}
+					}
+				}
+			}
+		},
+	)
+
+	srv.srv = httptest.NewServer(m)
+
+	u, err := url.Parse(srv.srv.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv.Endpoint = u.Host
+
+	testhelpers.WaitForEndpoint(srv.Endpoint)
+
+	return srv
+}
+
+func (m *MockServer) Close() {
+	m.srv.Close()
+}
diff --git a/client/internal/receiver.go b/client/internal/receiver.go
new file mode 100644
index 0000000..4f36160
--- /dev/null
+++ b/client/internal/receiver.go
@@ -0,0 +1,168 @@
+package internal
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/gorilla/websocket"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+// Receiver implements the client's receiving portion of OpAMP protocol.
+type Receiver struct {
+	conn      *websocket.Conn
+	logger    types.Logger
+	sender    *Sender
+	callbacks types.Callbacks
+}
+
+func NewReceiver(logger types.Logger, callbacks types.Callbacks, conn *websocket.Conn, sender *Sender) *Receiver {
+	return &Receiver{
+		conn:      conn,
+		logger:    logger,
+		sender:    sender,
+		callbacks: callbacks,
+	}
+}
+
+func (r *Receiver) ReceiverLoop(ctx context.Context) {
+	runContext, cancelFunc := context.WithCancel(ctx)
+
+out:
+	for {
+		var message protobufs.ServerToAgent
+		if err := r.receiveMessage(&message); err != nil {
+			if ctx.Err() == nil && !websocket.IsCloseError(err, websocket.CloseNormalClosure) {
+				r.logger.Errorf("Unexpected error while receiving: %v", err)
+			}
+			break out
+		} else {
+			r.processReceivedMessage(runContext, &message)
+		}
+	}
+
+	cancelFunc()
+}
+
+func (r *Receiver) receiveMessage(msg *protobufs.ServerToAgent) error {
+	_, bytes, err := r.conn.ReadMessage()
+	if err != nil {
+		return err
+	}
+	err = proto.Unmarshal(bytes, msg)
+	if err != nil {
+		return fmt.Errorf("cannot decode received message: %w", err)
+	}
+	return err
+}
+
+func (r *Receiver) processReceivedMessage(ctx context.Context, p *protobufs.ServerToAgent) {
+	data := p.GetDataForAgent()
+	if data != nil {
+		r.rcvDataForAgent(ctx, data)
+		return
+	}
+
+	err := p.GetErrorResponse()
+	if err != nil {
+		r.processErrorResponse(err)
+	}
+}
+
+func (r *Receiver) rcvDataForAgent(ctx context.Context, data *protobufs.DataForAgent) {
+	if r.callbacks != nil {
+		reportStatus := r.rcvRemoteConfig(ctx, data.RemoteConfig)
+
+		r.rcvConnectionSettings(ctx, data.ConnectionSettings)
+		r.rcvAddonsAvailable(data.AddonsAvailable)
+
+		if reportStatus {
+			r.sender.ScheduleSend()
+		}
+	}
+}
+
+func (r *Receiver) rcvRemoteConfig(ctx context.Context, config *protobufs.AgentRemoteConfig) (reportStatus bool) {
+	effective, err := r.callbacks.OnRemoteConfig(ctx, config)
+	if err == nil {
+		r.sender.UpdateStatus(func(statusReport *protobufs.StatusReport) {
+			statusReport.EffectiveConfig = effective
+		})
+		if effective != nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *Receiver) rcvConnectionSettings(ctx context.Context, settings *protobufs.ConnectionSettingsOffers) {
+	if settings == nil {
+		return
+	}
+
+	if settings.Opamp != nil {
+		err := r.callbacks.OnOpampConnectionSettings(ctx, settings.Opamp)
+		if err == nil {
+			// TODO: verify connection using new settings.
+			r.callbacks.OnOpampConnectionSettingsAccepted(settings.Opamp)
+		}
+	}
+
+	r.rcvOwnTelemetryConnectionSettings(
+		ctx,
+		settings.OwnMetrics,
+		types.OwnMetrics,
+		r.callbacks.OnOwnTelemetryConnectionSettings,
+	)
+
+	r.rcvOwnTelemetryConnectionSettings(
+		ctx,
+		settings.OwnTraces,
+		types.OwnTraces,
+		r.callbacks.OnOwnTelemetryConnectionSettings,
+	)
+
+	r.rcvOwnTelemetryConnectionSettings(
+		ctx,
+		settings.OwnLogs,
+		types.OwnLogs,
+		r.callbacks.OnOwnTelemetryConnectionSettings,
+	)
+
+	for name, s := range settings.OtherConnections {
+		r.rcvOtherConnectionSettings(ctx, s, name, r.callbacks.OnOtherConnectionSettings)
+	}
+}
+
+func (r *Receiver) rcvOwnTelemetryConnectionSettings(
+	ctx context.Context,
+	settings *protobufs.ConnectionSettings,
+	telemetryType types.OwnTelemetryType,
+	callback func(ctx context.Context, telemetryType types.OwnTelemetryType, settings *protobufs.ConnectionSettings) error,
+) {
+	if settings != nil {
+		callback(ctx, telemetryType, settings)
+	}
+}
+
+func (r *Receiver) rcvOtherConnectionSettings(
+	ctx context.Context,
+	settings *protobufs.ConnectionSettings,
+	name string,
+	callback func(ctx context.Context, name string, settings *protobufs.ConnectionSettings) error,
+) {
+	if settings != nil {
+		callback(ctx, name, settings)
+	}
+}
+
+func (r *Receiver) processErrorResponse(body *protobufs.ServerErrorResponse) {
+	// TODO: implement this.
+}
+
+func (r *Receiver) rcvAddonsAvailable(addons *protobufs.AddonsAvailable) {
+	// TODO: implement this.
+}
diff --git a/client/internal/sender.go b/client/internal/sender.go
new file mode 100644
index 0000000..ac5a1f4
--- /dev/null
+++ b/client/internal/sender.go
@@ -0,0 +1,142 @@
+package internal
+
+import (
+	"context"
+	"sync"
+
+	"github.com/gorilla/websocket"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+// Sender implements the client's sending portion of OpAMP protocol.
+type Sender struct {
+	instanceUid string
+	conn        *websocket.Conn
+
+	logger types.Logger
+
+	// Indicates that there are pending messages to send.
+	hasMessages chan struct{}
+
+	// The next status report to send.
+	statusReport protobufs.StatusReport
+	// Indicates that the status report is pending to be sent.
+	statusReportPending bool
+	// Mutex to protect the above 2 fields.
+	statusReportMutex sync.Mutex
+
+	// Indicates that the sender has fully stopped.
+	stopped chan struct{}
+}
+
+func NewSender(logger types.Logger) *Sender {
+	return &Sender{
+		logger:      logger,
+		hasMessages: make(chan struct{}, 1),
+	}
+}
+
+// Start the sender and send the first status report that was set via UpdateStatus()
+// earlier. To stop the Sender cancel the ctx.
+func (s *Sender) Start(ctx context.Context, instanceUid string, conn *websocket.Conn) error {
+	s.conn = conn
+	s.instanceUid = instanceUid
+	err := s.sendStatusReport()
+
+	// Run the sender in the background.
+	s.stopped = make(chan struct{})
+	go s.run(ctx)
+
+	return err
+}
+
+// WaitToStop blocks until the sender is stopped. To stop the sender cancel the context
+// that was passed to Start().
+func (s *Sender) WaitToStop() {
+	<-s.stopped
+}
+
+// UpdateStatus applies the specified modifier function to the status report that
+// will be sent next and marks the status report as pending to be sent.
+func (s *Sender) UpdateStatus(modifier func(statusReport *protobufs.StatusReport)) {
+	s.statusReportMutex.Lock()
+	modifier(&s.statusReport)
+	s.statusReportPending = true
+	s.statusReportMutex.Unlock()
+}
+
+// ScheduleSend signals to the sending goroutine to send the current status report
+// if it is pending. If there is no pending status report (e.g. status report was
+// already sent and "pending" flag is reset) then status report will not be sent.
+func (s *Sender) ScheduleSend() {
+	select {
+	case s.hasMessages <- struct{}{}:
+	default:
+		break
+	}
+}
+
+func (s *Sender) run(ctx context.Context) {
+out:
+	for {
+		select {
+		case <-s.hasMessages:
+			s.sendMessages()
+
+		case <-ctx.Done():
+			break out
+		}
+	}
+
+	close(s.stopped)
+}
+
+func (s *Sender) sendMessages() {
+	s.sendStatusReport()
+}
+
+func (s *Sender) sendStatusReport() error {
+	var statusReport *protobufs.StatusReport
+	s.statusReportMutex.Lock()
+	if s.statusReportPending {
+		// Clone the statusReport to have a copy for sending and avoid blocking
+		// future updates to statusReport field.
+		statusReport = proto.Clone(&s.statusReport).(*protobufs.StatusReport)
+		s.statusReportPending = false
+
+		// Reset fields that we do not have to send unless they change before the
+		// next report after this one.
+		s.statusReport.RemoteConfigStatus = nil
+		s.statusReport.EffectiveConfig = nil
+		s.statusReport.AgentDescription = nil
+	}
+	s.statusReportMutex.Unlock()
+
+	if statusReport != nil {
+		msg := protobufs.AgentToServer{
+			InstanceUid: s.instanceUid,
+			Body: &protobufs.AgentToServer_StatusReport{
+				StatusReport: statusReport,
+			},
+		}
+		return s.sendMessage(&msg)
+	}
+	return nil
+}
+
+func (s *Sender) sendMessage(msg *protobufs.AgentToServer) error {
+	data, err := proto.Marshal(msg)
+	if err != nil {
+		s.logger.Errorf("Cannot marshal data: %v", err)
+		return err
+	}
+	err = s.conn.WriteMessage(websocket.BinaryMessage, data)
+	if err != nil {
+		s.logger.Errorf("Cannot send: %v", err)
+		// TODO: propagate error back to Client and reconnect.
+	}
+	return err
+}
diff --git a/client/noplogger.go b/client/noplogger.go
new file mode 100644
index 0000000..61eea18
--- /dev/null
+++ b/client/noplogger.go
@@ -0,0 +1,6 @@
+package client
+
+type nopLogger struct{}
+
+func (l *nopLogger) Debugf(format string, v ...interface{}) {}
+func (l *nopLogger) Errorf(format string, v ...interface{}) {}
diff --git a/client/types/logger.go b/client/types/logger.go
new file mode 100644
index 0000000..c258b93
--- /dev/null
+++ b/client/types/logger.go
@@ -0,0 +1,6 @@
+package types
+
+type Logger interface {
+	Debugf(format string, v ...interface{})
+	Errorf(format string, v ...interface{})
+}
diff --git a/go.mod b/go.mod
index d693ee7..f934e76 100644
--- a/go.mod
+++ b/go.mod
@@ -2,4 +2,17 @@ module github.com/open-telemetry/opamp-go
 
 go 1.17
 
-require google.golang.org/protobuf v1.27.1
+require (
+	github.com/cenkalti/backoff/v4 v4.1.2
+	github.com/gorilla/websocket v1.4.2
+	github.com/oklog/ulid/v2 v2.0.2
+	github.com/stretchr/testify v1.7.0
+	google.golang.org/protobuf v1.27.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
+)
diff --git a/go.sum b/go.sum
index 03b1917..a701798 100644
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,28 @@
+github.com/cenkalti/backoff/v4 v4.1.2 h1:6Yo7N8UP2K6LWZnW94DLVSSrbobcWdVzAYOisuDPIFo=
+github.com/cenkalti/backoff/v4 v4.1.2/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/oklog/ulid/v2 v2.0.2 h1:r4fFzBm+bv0wNKNh5eXTwU7i85y5x+uwkxCUTNVQqLc=
+github.com/oklog/ulid/v2 v2.0.2/go.mod h1:mtBL0Qe/0HAx6/a4Z30qxVIAL1eQDweXq5lxOEiwQ68=
+github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
+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.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/server/server.go b/server/server.go
new file mode 100644
index 0000000..e04c4fa
--- /dev/null
+++ b/server/server.go
@@ -0,0 +1,49 @@
+package server
+
+import (
+	"context"
+	"crypto/tls"
+	"net/http"
+
+	"github.com/open-telemetry/opamp-go/server/types"
+)
+
+type Settings struct {
+	// Callbacks that the server will call after successful Attach/Start.
+	Callbacks types.Callbacks
+}
+
+type StartSettings struct {
+	Settings
+
+	// ListenEndpoint specifies the endpoint to listen on, e.g. "127.0.0.1:4320"
+	ListenEndpoint string
+
+	// Server's TLS configuration.
+	TLSConfig *tls.Config
+}
+
+type HTTPHandlerFunc func(http.ResponseWriter, *http.Request)
+
+type OpAMPServer interface {
+	// Attach prepares the OpAMP Server to begin handling requests from an existing
+	// http.Server. The returned HTTPHandlerFunc should be added as a handler to the
+	// desired http.Server by the caller and the http.Server should be started by
+	// the caller after that.
+	// For example:
+	//   handler, _ := server.Attach()
+	//   mux := http.NewServeMux()
+	//   mux.HandleFunc("/opamp", handler)
+	//   httpSrv := &http.Server{Handler:mux,Addr:"127.0.0.1:4320"}
+	//   httpSrv.ListenAndServe()
+	Attach(settings Settings) (HTTPHandlerFunc, error)
+
+	// Start an OpAMP Server and begin accepting connections. Starts its own http.Server
+	// using provided settings. This should block until the http.Server is ready to
+	// accept connections.
+	Start(settings StartSettings) error
+
+	// Stop accepting new connections and close all current connections. This should
+	// block until all connections are closed.
+	Stop(ctx context.Context) error
+}
diff --git a/server/types/callbacks.go b/server/types/callbacks.go
new file mode 100644
index 0000000..dfd23b5
--- /dev/null
+++ b/server/types/callbacks.go
@@ -0,0 +1,50 @@
+package types
+
+import (
+	"net/http"
+
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+// ConnectionResponse is the return type of the OnConnecting callback.
+type ConnectionResponse struct {
+	Accept             bool
+	HTTPStatusCode     int
+	HTTPResponseHeader map[string]string
+}
+
+type Callbacks interface {
+	// The following callbacks will never be called concurrently for the same
+	// connection. They may be called concurrently for different connections.
+
+	// OnConnecting is called when there is a new incoming connection.
+	// The handler can examine the request and either accept or reject the connection.
+	// To accept:
+	//   Return ConnectionResponse with Accept=true.
+	//   HTTPStatusCode and HTTPResponseHeader are ignored.
+	//
+	// To reject:
+	//   Return ConnectionResponse with Accept=false. HTTPStatusCode MUST be set to
+	//   non-zero value to indicate the rejection reason (typically 401, 429 or 503).
+	//   HTTPResponseHeader may be optionally set (e.g. "Retry-After: 30").
+	OnConnecting(request *http.Request) ConnectionResponse
+
+	// OnConnected is called when the WebSocket connection is successfully established
+	// after OnConnecting() returns and the HTTP connection is upgraded to WebSocket.
+	OnConnected(conn Connection)
+
+	// OnMessage is called when a message is received from the connection. Can happen
+	// only after OnConnected().
+	OnMessage(conn Connection, message *protobufs.AgentToServer)
+
+	// OnDisconnect is called when a Disconnect message is received from the client
+	// identified by instanceUid through the specified connection. Note that this is
+	// not the same as disconnecting the connection, the Disconnect is a message that
+	// the clients should send before the actual disconnection.
+	OnDisconnect(conn Connection, instanceUid string)
+
+	// OnConnectionClose is called when the WebSocket connection is closed.
+	// Typically, preceded by OnDisconnect() unless the client misbehaves or the
+	// connection is lost.
+	OnConnectionClose(conn Connection)
+}
diff --git a/server/types/connection.go b/server/types/connection.go
new file mode 100644
index 0000000..d4dba6b
--- /dev/null
+++ b/server/types/connection.go
@@ -0,0 +1,16 @@
+package types
+
+import (
+	"context"
+
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+// Connection represents one OpAMP WebSocket connections.
+// The implementation MUST be a comparable type so that it can be used as a map key.
+type Connection interface {
+	// Send a message. Should not be called concurrently for the same Connection instance.
+	// Blocks until the message is sent.
+	// Should return as soon as possible if the ctx is cancelled.
+	Send(ctx context.Context, message *protobufs.ServerToAgent) error
+}


Test patch:
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
new file mode 100644
index 0000000..b53308a
--- /dev/null
+++ b/client/clientimpl_test.go
@@ -0,0 +1,444 @@
+package client
+
+import (
+	"context"
+	"math/rand"
+	"net/http"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/oklog/ulid/v2"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/client/internal"
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+	"github.com/open-telemetry/opamp-go/internal/testhelpers"
+)
+
+func TestConnectInvalidURL(t *testing.T) {
+	settings := StartSettings{
+		OpAMPServerURL: ":not a url",
+	}
+
+	client := New(nil)
+	err := client.Start(settings)
+	assert.Error(t, err)
+}
+
+func eventually(t *testing.T, f func() bool) {
+	assert.Eventually(t, f, 5*time.Second, 10*time.Millisecond)
+}
+
+func prepareClient(settings StartSettings) *client {
+	// Autogenerate instance id.
+	entropy := ulid.Monotonic(rand.New(rand.NewSource(99)), 0)
+	settings.InstanceUid = ulid.MustNew(ulid.Timestamp(time.Now()), entropy).String()
+
+	return New(nil)
+}
+
+func startClient(t *testing.T, settings StartSettings) *client {
+	client := prepareClient(settings)
+	err := client.Start(settings)
+	assert.NoError(t, err)
+	return client
+}
+
+func TestConnectNoServer(t *testing.T) {
+	settings := StartSettings{
+		OpAMPServerURL: "ws://" + testhelpers.GetAvailableLocalAddress(),
+	}
+
+	client := startClient(t, settings)
+
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestOnConnectFail(t *testing.T) {
+	var connectErr atomic.Value
+	settings := StartSettings{
+		OpAMPServerURL: "ws://" + testhelpers.GetAvailableLocalAddress(),
+		Callbacks: internal.CallbacksStruct{
+			OnConnectFailedFunc: func(err error) {
+				connectErr.Store(err)
+			},
+		},
+	}
+
+	client := startClient(t, settings)
+
+	eventually(t, func() bool { return connectErr.Load() != nil })
+
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestStartStarted(t *testing.T) {
+	settings := StartSettings{
+		OpAMPServerURL: "ws://" + testhelpers.GetAvailableLocalAddress(),
+	}
+
+	client := startClient(t, settings)
+
+	err := client.Start(settings)
+	assert.Error(t, err)
+
+	err = client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestStopWithoutStart(t *testing.T) {
+	client := New(nil)
+	err := client.Stop(context.Background())
+	assert.Error(t, err)
+}
+
+func TestStopCancellation(t *testing.T) {
+	settings := StartSettings{
+		OpAMPServerURL: "ws://" + testhelpers.GetAvailableLocalAddress(),
+	}
+	client := startClient(t, settings)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	err := client.Stop(ctx)
+	assert.ErrorIs(t, err, context.Canceled)
+}
+
+func TestConnectWithServer(t *testing.T) {
+	// Start a server.
+	srv := internal.StartMockServer(t)
+
+	// Start a client.
+	var connected int64
+	settings := StartSettings{
+		Callbacks: internal.CallbacksStruct{
+			OnConnectFunc: func() {
+				atomic.StoreInt64(&connected, 1)
+			},
+		},
+	}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := startClient(t, settings)
+
+	// Wait for connection to be established.
+	eventually(t, func() bool { return atomic.LoadInt64(&connected) != 0 })
+
+	// Shutdown the server.
+	srv.Close()
+
+	// Shutdown the client.
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestConnectWithServer503(t *testing.T) {
+	// Start a server.
+	var connectionAttempts int64
+	srv := internal.StartMockServer(t)
+	srv.OnRequest = func(w http.ResponseWriter, r *http.Request) {
+		atomic.StoreInt64(&connectionAttempts, 1)
+
+		// Always respond with an error to the client.
+		w.Header().Set(retryAfterHTTPHeader, "30")
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}
+
+	// Start a client.
+	var clientConnected int64
+	var connectErr atomic.Value
+	settings := StartSettings{
+		Callbacks: internal.CallbacksStruct{
+			OnConnectFunc: func() {
+				atomic.StoreInt64(&clientConnected, 1)
+				assert.Fail(t, "Client should not be able to connect")
+			},
+			OnConnectFailedFunc: func(err error) {
+				connectErr.Store(err)
+			},
+		},
+	}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := startClient(t, settings)
+
+	// Wait for connection to fail.
+	eventually(t, func() bool { return connectErr.Load() != nil })
+
+	assert.EqualValues(t, 1, atomic.LoadInt64(&connectionAttempts))
+	assert.EqualValues(t, 0, atomic.LoadInt64(&clientConnected))
+
+	// Shutdown the server.
+	srv.Close()
+	client.Stop(context.Background())
+}
+
+func TestConnectWithHeader(t *testing.T) {
+	// Start a server.
+	srv := internal.StartMockServer(t)
+	var conn atomic.Value
+	srv.OnConnect = func(r *http.Request, c *websocket.Conn) {
+		authHdr := r.Header.Get("Authorization")
+		assert.EqualValues(t, "Bearer 12345678", authHdr)
+		conn.Store(c)
+	}
+
+	// Start a client.
+	settings := StartSettings{
+		OpAMPServerURL:      "ws://" + srv.Endpoint,
+		AuthorizationHeader: "Bearer 12345678",
+	}
+	client := startClient(t, settings)
+
+	// Wait for connection to be established.
+	eventually(t, func() bool { return conn.Load() != nil })
+
+	// Shutdown the server and the client.
+	srv.Close()
+	client.Stop(context.Background())
+}
+
+func TestDisconnectByServer(t *testing.T) {
+	// Start a server.
+	srv := internal.StartMockServer(t)
+	var conn atomic.Value
+	srv.OnConnect = func(r *http.Request, c *websocket.Conn) {
+		conn.Store(c)
+	}
+
+	// Start a client.
+	var connected int64
+	var connectErr atomic.Value
+	settings := StartSettings{
+		Callbacks: internal.CallbacksStruct{
+			OnConnectFunc: func() {
+				atomic.StoreInt64(&connected, 1)
+			},
+			OnConnectFailedFunc: func(err error) {
+				connectErr.Store(err)
+			},
+		},
+	}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := startClient(t, settings)
+
+	// Wait for connection to be established.
+	eventually(t, func() bool { return conn.Load() != nil })
+	assert.True(t, connectErr.Load() == nil)
+
+	// Close the server and forcefully disconnect.
+	srv.Close()
+	conn.Load().(*websocket.Conn).Close()
+
+	// The client must retry and must fail now.
+	eventually(t, func() bool { return connectErr.Load() != nil })
+
+	// Stop the client.
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestFirstStatusReport(t *testing.T) {
+	remoteConfig := &protobufs.AgentRemoteConfig{
+		Config: &protobufs.AgentConfigMap{
+			ConfigMap: map[string]*protobufs.AgentConfigFile{},
+		},
+		ConfigHash: []byte{1, 2, 3, 4},
+	}
+
+	// Start a server.
+	srv := internal.StartMockServer(t)
+	srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+		return &protobufs.ServerToAgent{
+			InstanceUid: msg.InstanceUid,
+			Body: &protobufs.ServerToAgent_DataForAgent{
+				DataForAgent: &protobufs.DataForAgent{
+					RemoteConfig: remoteConfig,
+				},
+			},
+		}
+	}
+
+	// Start a client.
+	var connected, remoteConfigReceived int64
+	settings := StartSettings{
+		Callbacks: internal.CallbacksStruct{
+			OnConnectFunc: func() {
+				atomic.AddInt64(&connected, 1)
+			},
+			OnRemoteConfigFunc: func(
+				ctx context.Context, config *protobufs.AgentRemoteConfig,
+			) (*protobufs.EffectiveConfig, error) {
+				// Verify that the client received exactly the remote config that
+				// the server sent.
+				assert.True(t, proto.Equal(remoteConfig, config))
+				atomic.AddInt64(&remoteConfigReceived, 1)
+				return &protobufs.EffectiveConfig{
+					ConfigMap: remoteConfig.Config,
+				}, nil
+			},
+		},
+	}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := startClient(t, settings)
+
+	// Wait for connection to be established.
+	eventually(t, func() bool { return atomic.LoadInt64(&connected) != 0 })
+
+	// Wait to receive remote config.
+	eventually(t, func() bool { return atomic.LoadInt64(&remoteConfigReceived) != 0 })
+
+	// Shutdown the server.
+	srv.Close()
+
+	// Shutdown the client.
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestSetEffectiveConfig(t *testing.T) {
+	// Start a server.
+	srv := internal.StartMockServer(t)
+	var rcvConfig atomic.Value
+	srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+		if statusReport := msg.GetStatusReport(); statusReport != nil {
+			if statusReport.EffectiveConfig != nil {
+				rcvConfig.Store(statusReport.EffectiveConfig)
+			}
+		}
+		return nil
+	}
+
+	// Start a client.
+	settings := StartSettings{}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := prepareClient(settings)
+
+	sendConfig := &protobufs.EffectiveConfig{
+		ConfigMap: &protobufs.AgentConfigMap{
+			ConfigMap: map[string]*protobufs.AgentConfigFile{
+				"key": {},
+			},
+		},
+	}
+
+	// Set before start. It should be delivered immediately after Start().
+	client.SetEffectiveConfig(sendConfig)
+
+	assert.NoError(t, client.Start(settings))
+
+	// Verify it is delivered.
+	eventually(t, func() bool { return proto.Equal(sendConfig, rcvConfig.Load().(*protobufs.EffectiveConfig)) })
+
+	// Now change again.
+	sendConfig.ConfigMap.ConfigMap["key2"] = &protobufs.AgentConfigFile{}
+	client.SetEffectiveConfig(sendConfig)
+
+	// Verify change is delivered.
+	eventually(t, func() bool { return proto.Equal(sendConfig, rcvConfig.Load().(*protobufs.EffectiveConfig)) })
+
+	// Shutdown the server.
+	srv.Close()
+
+	// Shutdown the client.
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestConnectionSettings(t *testing.T) {
+	hash := []byte{1, 2, 3}
+	opampSettings := &protobufs.ConnectionSettings{DestinationEndpoint: "http://opamp.com"}
+	metricsSettings := &protobufs.ConnectionSettings{DestinationEndpoint: "http://metrics.com"}
+	tracesSettings := &protobufs.ConnectionSettings{DestinationEndpoint: "http://traces.com"}
+	logsSettings := &protobufs.ConnectionSettings{DestinationEndpoint: "http://logs.com"}
+	otherSettings := &protobufs.ConnectionSettings{DestinationEndpoint: "http://other.com"}
+
+	var rcvStatus int64
+	// Start a server.
+	srv := internal.StartMockServer(t)
+	srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+		if statusReport := msg.GetStatusReport(); statusReport != nil {
+			atomic.AddInt64(&rcvStatus, 1)
+
+			return &protobufs.ServerToAgent{
+				Body: &protobufs.ServerToAgent_DataForAgent{
+					DataForAgent: &protobufs.DataForAgent{
+						ConnectionSettings: &protobufs.ConnectionSettingsOffers{
+							Hash:       hash,
+							Opamp:      opampSettings,
+							OwnMetrics: metricsSettings,
+							OwnTraces:  tracesSettings,
+							OwnLogs:    logsSettings,
+							OtherConnections: map[string]*protobufs.ConnectionSettings{
+								"other": otherSettings,
+							},
+						},
+					},
+				},
+			}
+		}
+		return nil
+	}
+
+	var gotOpampSettings int64
+	var gotOwnSettings int64
+	var gotOtherSettings int64
+
+	// Start a client.
+	settings := StartSettings{
+		Callbacks: internal.CallbacksStruct{
+			OnOpampConnectionSettingsFunc: func(
+				ctx context.Context, settings *protobufs.ConnectionSettings,
+			) error {
+				assert.True(t, proto.Equal(opampSettings, settings))
+				atomic.AddInt64(&gotOpampSettings, 1)
+				return nil
+			},
+
+			OnOwnTelemetryConnectionSettingsFunc: func(
+				ctx context.Context, telemetryType types.OwnTelemetryType,
+				settings *protobufs.ConnectionSettings,
+			) error {
+				switch telemetryType {
+				case types.OwnMetrics:
+					assert.True(t, proto.Equal(metricsSettings, settings))
+				case types.OwnTraces:
+					assert.True(t, proto.Equal(tracesSettings, settings))
+				case types.OwnLogs:
+					assert.True(t, proto.Equal(logsSettings, settings))
+				}
+				atomic.AddInt64(&gotOwnSettings, 1)
+				return nil
+			},
+
+			OnOtherConnectionSettingsFunc: func(
+				ctx context.Context, name string, settings *protobufs.ConnectionSettings,
+			) error {
+				assert.EqualValues(t, "other", name)
+				assert.True(t, proto.Equal(otherSettings, settings))
+				atomic.AddInt64(&gotOtherSettings, 1)
+				return nil
+			},
+		},
+	}
+	settings.OpAMPServerURL = "ws://" + srv.Endpoint
+	client := prepareClient(settings)
+
+	assert.NoError(t, client.Start(settings))
+
+	eventually(t, func() bool { return atomic.LoadInt64(&gotOpampSettings) == 1 })
+	eventually(t, func() bool { return atomic.LoadInt64(&gotOwnSettings) == 3 })
+	eventually(t, func() bool { return atomic.LoadInt64(&gotOtherSettings) == 1 })
+	eventually(t, func() bool { return atomic.LoadInt64(&rcvStatus) == 1 })
+
+	// Shutdown the server.
+	srv.Close()
+
+	// Shutdown the client.
+	err := client.Stop(context.Background())
+	assert.NoError(t, err)
+}
diff --git a/internal/testhelpers/nethelpers.go b/internal/testhelpers/nethelpers.go
new file mode 100644
index 0000000..7ad7048
--- /dev/null
+++ b/internal/testhelpers/nethelpers.go
@@ -0,0 +1,70 @@
+package testhelpers
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// GetAvailableLocalAddress finds an available local port and returns an endpoint
+// describing it. The port is available for opening when this function returns
+// provided that there is no race by some other code to grab the same port
+// immediately.
+func GetAvailableLocalAddress() string {
+	ln, err := net.Listen("tcp", "127.0.0.1:")
+	if err != nil {
+		log.Fatalf("failed to get a free local port: %v", err)
+	}
+	// There is a possible race if something else takes this same port before
+	// the test uses it, however, that is unlikely in practice.
+	defer ln.Close()
+	return ln.Addr().String()
+}
+
+func waitForPortToListen(port int) error {
+	totalDuration := 5 * time.Second
+	wait := 10 * time.Millisecond
+	address := fmt.Sprintf("127.0.0.1:%d", port)
+
+	ticker := time.NewTicker(wait)
+	defer ticker.Stop()
+
+	timeout := time.After(totalDuration)
+
+	for {
+		select {
+		case <-ticker.C:
+			conn, err := net.Dial("tcp", address)
+			if err == nil && conn != nil {
+				conn.Close()
+				return nil
+			}
+
+		case <-timeout:
+			return fmt.Errorf("failed to wait for port %d", port)
+		}
+	}
+}
+
+// HostPortFromAddr extracts host and port from a network address
+func HostPortFromAddr(endpoint string) (host string, port int, err error) {
+	sepIndex := strings.LastIndex(endpoint, ":")
+	if sepIndex < 0 {
+		return "", -1, errors.New("failed to parse host:port")
+	}
+	host, portStr := endpoint[:sepIndex], endpoint[sepIndex+1:]
+	port, err = strconv.Atoi(portStr)
+	return host, port, err
+}
+
+func WaitForEndpoint(endpoint string) {
+	_, port, err := HostPortFromAddr(endpoint)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	waitForPortToListen(port)
+}