Repository: open-telemetry/opamp-go
Pull request #353 (2025-02-20 14:14:06)

Improve contributor guide
This is part of https://github.com/open-telemetry/sig-contributor-experience/issues/31

The goal is to create a consistent guide that makes it easy for first time conributors to contribute to the repo.

### Changes:

-Introduction section to welcome first time contributors
-Local Run/Build - Updated this section with existing information but made it clearer for developers
-Further Help - Added this section to direct contributors to slack or create issues for more help.

Feeback is welcome


Patch:
diff --git a/.gitignore b/.gitignore
index 1e6c714..8311ed7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -18,3 +18,6 @@
 coverage.out
 coverage.txt
 coverage.html
+
+# Project tools
+.tools/
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 4f803f3..03da2ed 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,22 +1,45 @@
 # Contributing to opamp-go library
 
+Welcome, and thank you for your interest in contributing to the opamp-go library!
+
+Whether you're submitting a bug fix, implementing a new feature, or updating documentation, we appreciate your effort in making opamp-go better! If you're new to the project, don’t hesitate to ask for help — we value collaboration and learning.
+
+Before getting started, please ensure you’re familiar with the contribution process by reviewing this guide. Also, check out the [OpenTelemetry contributing guide](https://github.com/open-telemetry/community/blob/main/guides/contributor/README.md) for general contribution guidelines.
+
+## Pre-requisites
+
+* Ensure you have Go installed (check the required version in [go.mod](https://github.com/open-telemetry/opamp-go/blob/main/go.mod)).
+
+* Install Docker, which is required for generating protobuf files.
+
 ## Generate Protobuf Go Files
 
-1. Make sure `opamp-go/internal/opamp-spec` submodule files exist. You can do this
-either by cloning the `opamp-go` repo with submodules the first time, e.g. 
-`git clone --recurse-submodules git@github.com:open-telemetry/opamp-go.git`.
-Alternatively, if you already cloned `opamp-go` without the submodule then
-execute `git submodule update --init` to fetch all files for `opamp-spec` submodule.
-Note that `opamp-spec` submodule requires ssh git cloning access to github and won't
-work if you only have https access.
+To set up and run opamp-go locally, follow these steps:
+
+Clone the repository with submodules:
+
+```
+git clone --recurse-submodules git@github.com:open-telemetry/opamp-go.git
+cd opamp-go
+```
 
-2. Make sure you have Docker installed and running.
+If you've already cloned the repository without submodules, initialize them with:
 
-3. Run `make gen-proto`. This should compile `internal/proto/*.proto` files to 
-`internal/protobufs/*.pb.go` files.
+```
+git submodule update --init
+```
 
-Note that this is tested on linux/amd64 and darwin/amd64 and is known to not work
-on darwin/arm64 (M1/M2 Macs). Fixes are welcome.
+This will fetch all files for `opamp-spec` submodule. 
+Note that `opamp-spec` submodule requires ssh git cloning access to github and won't work if you only have https access.
+
+Generate the protobuf Go files:
+
+```
+make gen-proto
+```
+This should compile `internal/proto/*.proto` files to `internal/protobufs/*.pb.go` files.
+
+Note that this is tested on linux/amd64 and darwin/amd64 and is known to not work on darwin/arm64 (M1/M2 Macs). Fixes are welcome.
 
 # Releasing a new version of opamp-go
 
@@ -29,3 +52,13 @@ on darwin/arm64 (M1/M2 Macs). Fixes are welcome.
 4. Check `Set as a pre-release` as appropriate.
 
 5. `Publish release`
+
+## Further Help
+
+If you have any questions or run into issues while contributing, feel free to reach out to the OpenTelemetry community:
+
+* Slack: Join the CNCF [#otel-opamp](https://cloud-native.slack.com/archives/C02J58HR58R) Slack channel to connect with other contributors and get access to SIG meeting notes
+  
+* GitHub Issues & Discussions: Report bugs or propose features via [GitHub Issues](https://github.com/open-telemetry/opamp-go/issues).
+
+We’re excited to have you contribute and look forward to your ideas and improvements! 🚀
diff --git a/client/client.go b/client/client.go
index f65e824..ae054f3 100644
--- a/client/client.go
+++ b/client/client.go
@@ -9,7 +9,6 @@ import (
 
 // OpAMPClient is an interface representing the client side of the OpAMP protocol.
 type OpAMPClient interface {
-
 	// Start the client and begin attempts to connect to the Server. Once connection
 	// is established the client will attempt to maintain it by reconnecting if
 	// the connection is lost. All failed connection attempts will be reported via
diff --git a/client/internal/httpsender.go b/client/internal/httpsender.go
index a97e131..3573b1f 100644
--- a/client/internal/httpsender.go
+++ b/client/internal/httpsender.go
@@ -21,11 +21,15 @@ import (
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-const OpAMPPlainHTTPMethod = "POST"
-const defaultPollingIntervalMs = 30 * 1000 // default interval is 30 seconds.
+const (
+	OpAMPPlainHTTPMethod     = "POST"
+	defaultPollingIntervalMs = 30 * 1000 // default interval is 30 seconds.
+)
 
-const headerContentEncoding = "Content-Encoding"
-const encodingTypeGZip = "gzip"
+const (
+	headerContentEncoding = "Content-Encoding"
+	encodingTypeGZip      = "gzip"
+)
 
 type requestWrapper struct {
 	*http.Request
diff --git a/client/internal/mockserver.go b/client/internal/mockserver.go
index 2d88ac9..87a3b09 100644
--- a/client/internal/mockserver.go
+++ b/client/internal/mockserver.go
@@ -34,8 +34,10 @@ type MockServer struct {
 	enableCompression bool
 }
 
-const headerContentType = "Content-Type"
-const contentTypeProtobuf = "application/x-protobuf"
+const (
+	headerContentType   = "Content-Type"
+	contentTypeProtobuf = "application/x-protobuf"
+)
 
 func newMockServer(t *testing.T) (*MockServer, *http.ServeMux) {
 	srv := &MockServer{
@@ -136,7 +138,7 @@ func (m *MockServer) EnableCompression() {
 }
 
 func (m *MockServer) handleWebSocket(t *testing.T, w http.ResponseWriter, r *http.Request) {
-	var upgrader = websocket.Upgrader{
+	upgrader := websocket.Upgrader{
 		EnableCompression: m.enableCompression,
 	}
 
diff --git a/client/internal/packagessyncer.go b/client/internal/packagessyncer.go
index c02828d..ff6e7c0 100644
--- a/client/internal/packagessyncer.go
+++ b/client/internal/packagessyncer.go
@@ -47,7 +47,6 @@ func NewPackagesSyncer(
 
 // Sync performs the package syncing process.
 func (s *packagesSyncer) Sync(ctx context.Context) error {
-
 	defer func() {
 		close(s.doneCh)
 	}()
@@ -162,7 +161,6 @@ func (s *packagesSyncer) syncPackage(
 	pkgName string,
 	pkgAvail *protobufs.PackageAvailable,
 ) error {
-
 	status := s.statuses.Packages[pkgName]
 	if status == nil {
 		// This package has no status. Create one.
@@ -253,7 +251,8 @@ func (s *packagesSyncer) syncPackageFile(
 // shouldDownloadFile returns true if the file should be downloaded.
 func (s *packagesSyncer) shouldDownloadFile(ctx context.Context,
 	packageName string,
-	file *protobufs.DownloadableFile) (bool, error) {
+	file *protobufs.DownloadableFile,
+) (bool, error) {
 	fileContentHash, err := s.localState.FileContentHash(packageName)
 
 	if err != nil {
@@ -280,6 +279,12 @@ func (s *packagesSyncer) downloadFile(ctx context.Context, pkgName string, file
 		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
 	}
 
+	if file.Headers != nil {
+		for _, h := range file.Headers.Headers {
+			req.Header.Add(h.GetKey(), h.GetValue())
+		}
+	}
+
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
diff --git a/internal/certs.go b/internal/certs.go
index 6b20606..31e4980 100644
--- a/internal/certs.go
+++ b/internal/certs.go
@@ -82,7 +82,6 @@ func CreateServerTLSConfig(caCertPath, serverCertPath, serverKeyPath string) (*t
 }
 
 func CreateTLSCert(caCertPath, caKeyPath string) (*protobufs.TLSCertificate, error) {
-
 	// Load CA Cert.
 	caCertBytes, err := os.ReadFile(caCertPath)
 	if err != nil {
diff --git a/internal/examples/agent/agent/agent.go b/internal/examples/agent/agent/agent.go
index 7b5bfd0..38bb152 100644
--- a/internal/examples/agent/agent/agent.go
+++ b/internal/examples/agent/agent/agent.go
@@ -223,7 +223,7 @@ func (agent *Agent) updateAgentIdentity(ctx context.Context, instanceId uuid.UUI
 }
 
 func (agent *Agent) loadLocalConfig() {
-	var k = koanf.New(".")
+	k := koanf.New(".")
 	_ = k.Load(rawbytes.Provider([]byte(localConfig)), yaml.Parser())
 
 	effectiveConfigBytes, err := k.Marshal(yaml.Parser())
@@ -291,7 +291,7 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (conf
 	agent.logger.Debugf(context.Background(), "Received remote config from server, hash=%x.", config.ConfigHash)
 
 	// Begin with local config. We will later merge received configs on top of it.
-	var k = koanf.New(".")
+	k := koanf.New(".")
 	if err := k.Load(rawbytes.Provider([]byte(localConfig)), yaml.Parser()); err != nil {
 		return false, err
 	}
@@ -322,7 +322,7 @@ func (agent *Agent) applyRemoteConfig(config *protobufs.AgentRemoteConfig) (conf
 
 	// Merge received configs.
 	for _, item := range orderedConfigs {
-		var k2 = koanf.New(".")
+		k2 := koanf.New(".")
 		err := k2.Load(rawbytes.Provider(item.file.Body), yaml.Parser())
 		if err != nil {
 			return false, fmt.Errorf("cannot parse config named %s: %v", item.name, err)
diff --git a/internal/examples/agent/agent/metricreporter.go b/internal/examples/agent/agent/metricreporter.go
index 626ba53..7478307 100644
--- a/internal/examples/agent/agent/metricreporter.go
+++ b/internal/examples/agent/agent/metricreporter.go
@@ -48,7 +48,6 @@ func NewMetricReporter(
 	agentVersion string,
 	instanceId uuid.UUID,
 ) (*MetricReporter, error) {
-
 	// Check the destination credentials to make sure they look like a valid OTLP/HTTP
 	// destination.
 
@@ -168,7 +167,6 @@ func (reporter *MetricReporter) processMemoryPhysicalFunc(ctx context.Context, r
 }
 
 func (reporter *MetricReporter) sendMetrics() {
-
 	// Collect metrics every 5 seconds.
 	t := time.NewTicker(time.Second * 5)
 	ticks := int64(0)
diff --git a/internal/examples/server/certman/certman.go b/internal/examples/server/certman/certman.go
index 5f9be47..231d389 100644
--- a/internal/examples/server/certman/certman.go
+++ b/internal/examples/server/certman/certman.go
@@ -21,9 +21,11 @@ import (
 
 var logger = log.New(log.Default().Writer(), "[CertMan] ", log.Default().Flags()|log.Lmsgprefix|log.Lmicroseconds)
 
-var caCert *x509.Certificate
-var caPrivKey *rsa.PrivateKey
-var caCertBytes []byte
+var (
+	caCert      *x509.Certificate
+	caPrivKey   *rsa.PrivateKey
+	caCertBytes []byte
+)
 
 var loadCACertOnce sync.Once
 
diff --git a/internal/examples/server/uisrv/ui.go b/internal/examples/server/uisrv/ui.go
index f766b03..4d1d964 100644
--- a/internal/examples/server/uisrv/ui.go
+++ b/internal/examples/server/uisrv/ui.go
@@ -15,8 +15,10 @@ import (
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-var htmlDir string
-var srv *http.Server
+var (
+	htmlDir string
+	srv     *http.Server
+)
 
 var logger = log.New(log.Default().Writer(), "[UI] ", log.Default().Flags()|log.Lmsgprefix|log.Lmicroseconds)
 
@@ -44,7 +46,6 @@ func renderTemplate(w http.ResponseWriter, htmlTemplateFile string, data interfa
 		path.Join(htmlDir, "header.html"),
 		path.Join(htmlDir, htmlTemplateFile),
 	)
-
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
 		logger.Printf("Error parsing html template %s: %v", htmlTemplateFile, err)
diff --git a/internal/examples/supervisor/supervisor/commander/commander.go b/internal/examples/supervisor/supervisor/commander/commander.go
index bd2f1f6..cb5faa2 100644
--- a/internal/examples/supervisor/supervisor/commander/commander.go
+++ b/internal/examples/supervisor/supervisor/commander/commander.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -24,6 +25,9 @@ type Commander struct {
 	doneCh  chan struct{}
 	waitCh  chan struct{}
 	running int64
+
+	// Process should not be started while being stopped.
+	startStopMutex sync.RWMutex
 }
 
 func NewCommander(logger types.Logger, cfg *config.Agent, args ...string) (*Commander, error) {
@@ -41,6 +45,9 @@ func NewCommander(logger types.Logger, cfg *config.Agent, args ...string) (*Comm
 // Start the Agent and begin watching the process.
 // Agent's stdout and stderr are written to a file.
 func (c *Commander) Start(ctx context.Context) error {
+	c.startStopMutex.Lock()
+	defer c.startStopMutex.Unlock()
+
 	c.logger.Debugf(ctx, "Starting agent %s", c.cfg.Executable)
 
 	logFilePath := "agent.log"
@@ -116,6 +123,9 @@ func (c *Commander) IsRunning() bool {
 // and if the process does not finish kills it forcedly by sending SIGKILL.
 // Returns after the process is terminated.
 func (c *Commander) Stop(ctx context.Context) error {
+	c.startStopMutex.Lock()
+	defer c.startStopMutex.Unlock()
+
 	if c.cmd == nil || c.cmd.Process == nil {
 		// Not started, nothing to do.
 		return nil
diff --git a/internal/examples/supervisor/supervisor/healthchecker/healthchecker.go b/internal/examples/supervisor/supervisor/healthchecker/healthchecker.go
index 6a77106..861cec0 100644
--- a/internal/examples/supervisor/supervisor/healthchecker/healthchecker.go
+++ b/internal/examples/supervisor/supervisor/healthchecker/healthchecker.go
@@ -18,7 +18,6 @@ func NewHttpHealthChecker(endpoint string) *HttpHealthChecker {
 }
 
 func (h *HttpHealthChecker) Check(ctx context.Context) error {
-
 	req, err := http.NewRequestWithContext(ctx, "GET", h.endpoint, nil)
 	if err != nil {
 		return err
diff --git a/internal/examples/supervisor/supervisor/supervisor.go b/internal/examples/supervisor/supervisor/supervisor.go
index 18c014c..90c67f8 100644
--- a/internal/examples/supervisor/supervisor/supervisor.go
+++ b/internal/examples/supervisor/supervisor/supervisor.go
@@ -129,7 +129,6 @@ func (s *Supervisor) loadConfig() error {
 }
 
 func (s *Supervisor) startOpAMP() error {
-
 	s.opampClient = client.NewWebSocket(s.logger)
 
 	settings := types.StartSettings{
@@ -219,7 +218,6 @@ func (s *Supervisor) createAgentDescription() *protobufs.AgentDescription {
 }
 
 func (s *Supervisor) composeExtraLocalConfig() string {
-
 	return fmt.Sprintf(`
 service:
   telemetry:
@@ -331,7 +329,7 @@ service:
 // 1) the remote config from OpAMP Server, 2) the own metrics config section,
 // 3) the local override config that is hard-coded in the Supervisor.
 func (s *Supervisor) composeEffectiveConfig(ctx context.Context, config *protobufs.AgentRemoteConfig) (configChanged bool, err error) {
-	var k = koanf.New(".")
+	k := koanf.New(".")
 
 	// Begin with empty config. We will merge received configs on top of it.
 	if err := k.Load(rawbytes.Provider([]byte{}), yaml.Parser()); err != nil {
@@ -356,7 +354,7 @@ func (s *Supervisor) composeEffectiveConfig(ctx context.Context, config *protobu
 	// Merge received configs.
 	for _, name := range names {
 		item := config.Config.ConfigMap[name]
-		var k2 = koanf.New(".")
+		k2 := koanf.New(".")
 		err := k2.Load(rawbytes.Provider(item.Body), yaml.Parser())
 		if err != nil {
 			return false, fmt.Errorf("cannot parse config named %s: %v", name, err)
@@ -401,7 +399,6 @@ func (s *Supervisor) composeEffectiveConfig(ctx context.Context, config *protobu
 // Recalculate the Agent's effective config and if the config changes signal to the
 // background goroutine that the config needs to be applied to the Agent.
 func (s *Supervisor) recalcEffectiveConfig(ctx context.Context) (configChanged bool, err error) {
-
 	configChanged, err = s.composeEffectiveConfig(ctx, s.remoteConfig)
 	if err != nil {
 		s.logger.Errorf(ctx, "Error composing effective config. Ignoring received config: %v", err)
diff --git a/internal/tools/go.mod b/internal/tools/go.mod
index 33fe3e7..435a75c 100644
--- a/internal/tools/go.mod
+++ b/internal/tools/go.mod
@@ -1,13 +1,20 @@
 module github.com/open-telemetry/opamp-go/internal/tools
 
-go 1.22
+go 1.22.0
 
-require github.com/ory/go-acc v0.2.8
+toolchain go1.24.0
+
+require (
+	github.com/ory/go-acc v0.2.8
+	golang.org/x/tools v0.30.0
+	mvdan.cc/gofumpt v0.7.0
+)
 
 require (
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/dgraph-io/ristretto v0.0.2 // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/uuid v1.1.1 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
@@ -22,7 +29,9 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/subosito/gotenv v1.2.0 // indirect
-	golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 // indirect
+	golang.org/x/mod v0.23.0 // indirect
+	golang.org/x/sync v0.11.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.3.2 // indirect
 	gopkg.in/ini.v1 v1.57.0 // indirect
 	gopkg.in/yaml.v2 v2.3.0 // indirect
diff --git a/internal/tools/go.sum b/internal/tools/go.sum
index 28c436b..cb3c728 100644
--- a/internal/tools/go.sum
+++ b/internal/tools/go.sum
@@ -34,6 +34,8 @@ github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeME
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
+github.com/go-quicktest/qt v1.101.0 h1:O1K29Txy5P2OK0dGo59b7b0LR6wKfIhttaAhHUyn7eI=
+github.com/go-quicktest/qt v1.101.0/go.mod h1:14Bz/f7NwaXPtdYEgzsx46kqSxVwTbzVZsDC26tQJow=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
@@ -44,6 +46,8 @@ github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -66,11 +70,13 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzRKO2BQ4=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
@@ -103,6 +109,8 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
+github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
@@ -149,6 +157,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20200510223506-06a226fb4e37/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM=
+golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -159,6 +169,8 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -167,8 +179,9 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200124204421-9fbb57f87de9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 h1:OH54vjqzRWmbJ62fjuhxy7AxFFgoHN0/DPc/UrL8cAs=
 golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
@@ -178,6 +191,8 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.30.0 h1:BgcpHewrV5AUp2G9MebG4XPFI1E2W41zU1SaqVA9vJY=
+golang.org/x/tools v0.30.0/go.mod h1:c347cR/OJfw5TI+GfX7RUPNMdDRRbjvYTS0jPyvsVtY=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -198,3 +213,5 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+mvdan.cc/gofumpt v0.7.0 h1:bg91ttqXmi9y2xawvkuMXyvAA/1ZGJqYAEGjXuP0JXU=
+mvdan.cc/gofumpt v0.7.0/go.mod h1:txVFJy/Sc/mvaycET54pV8SW8gWxTlUuGHVEcncmNUo=
diff --git a/internal/tools/tools.go b/internal/tools/tools.go
index 99bc400..afd34fa 100644
--- a/internal/tools/tools.go
+++ b/internal/tools/tools.go
@@ -24,4 +24,6 @@ package tools
 
 import (
 	_ "github.com/ory/go-acc"
+	_ "golang.org/x/tools/cmd/goimports"
+	_ "mvdan.cc/gofumpt"
 )
diff --git a/makefile b/makefile
index d348a50..99a55ac 100644
--- a/makefile
+++ b/makefile
@@ -5,7 +5,20 @@ $(1)
 
 endef
 
-TOOLS_MOD_DIR := ./internal/tools
+GOCMD?= go
+
+# SRC_ROOT is the top of the source tree.
+SRC_ROOT := $(shell git rev-parse --show-toplevel)
+
+TOOLS_MOD_DIR   := $(SRC_ROOT)/internal/tools
+TOOLS_BIN_DIR   := $(SRC_ROOT)/.tools
+TOOLS_MOD_REGEX := "\s+_\s+\".*\""
+TOOLS_PKG_NAMES := $(shell grep -E $(TOOLS_MOD_REGEX) < $(TOOLS_MOD_DIR)/tools.go | tr -d " _\"" | grep -vE '/v[0-9]+$$')
+TOOLS_BIN_NAMES := $(addprefix $(TOOLS_BIN_DIR)/, $(notdir $(shell echo $(TOOLS_PKG_NAMES))))
+
+GOFUMPT      := $(TOOLS_BIN_DIR)/gofumpt
+GOIMPORTS    := $(TOOLS_BIN_DIR)/goimports
+GOACC	     := $(TOOLS_BIN_DIR)/go-acc
 
 # Find all .proto files.
 BASELINE_PROTO_FILES := $(wildcard internal/opamp-spec/proto/*.proto)
@@ -14,28 +27,28 @@ all: test build-examples
 
 .PHONY: test
 test:
-	go test -race ./...
-	cd internal/examples && go test -race ./...
+	$(GOCMD) test -race ./...
+	cd internal/examples && $(GOCMD) test -race ./...
 
 .PHONY: test-with-cover
-test-with-cover:
-	go-acc --output=coverage.out --ignore=protobufs ./...
+test-with-cover: $(GOACC)
+	$(GOACC) --output=coverage.out --ignore=protobufs ./...
 
 show-coverage: test-with-cover
 	# Show coverage as HTML in the default browser.
-	go tool cover -html=coverage.out
+	$(GOCMD) tool cover -html=coverage.out
 
 .PHONY: build-examples
 build-examples: build-example-agent build-example-supervisor build-example-server
 
 build-example-agent:
-	cd internal/examples && go build -o agent/bin/agent agent/main.go
+	cd internal/examples && $(GOCMD) build -o agent/bin/agent agent/main.go
 
 build-example-supervisor:
-	cd internal/examples && go build -o supervisor/bin/supervisor supervisor/main.go
+	cd internal/examples && $(GOCMD) build -o supervisor/bin/supervisor supervisor/main.go
 
 build-example-server:
-	cd internal/examples && go build -o server/bin/server server/main.go
+	cd internal/examples && $(GOCMD) build -o server/bin/server server/main.go
 
 run-examples: build-examples
 	cd internal/examples/server && ./bin/server &
@@ -61,18 +74,30 @@ gen-proto:
 
 	cp -R internal/proto/github.com/open-telemetry/opamp-go/protobufs/* protobufs/
 	rm -rf internal/proto/github.com/
+	$(MAKE) fmt
 
 .PHONY: gomoddownload
 gomoddownload:
-	go mod download
+	$(GOCMD) mod download
 
 .PHONY: install-tools
-install-tools:
-	cd $(TOOLS_MOD_DIR) && go install github.com/ory/go-acc
+install-tools: $(TOOLS_BIN_NAMES)
+
+$(TOOLS_BIN_DIR):
+	mkdir -p $@
+
+$(TOOLS_BIN_NAMES): $(TOOLS_BIN_DIR) $(TOOLS_MOD_DIR)/go.mod
+	cd $(TOOLS_MOD_DIR) && $(GOCMD) build -o $@ -trimpath $(filter %/$(notdir $@),$(TOOLS_PKG_NAMES))
 
 .PHONY: tidy
 tidy:
 	rm -fr go.sum
-	go mod tidy
-	cd internal/examples && rm -fr go.sum && go mod tidy
-	cd internal/tools && rm -fr go.sum && go mod tidy
+	$(GOCMD) mod tidy
+	cd internal/examples && rm -fr go.sum && $(GOCMD) mod tidy
+	cd $(TOOLS_MOD_DIR) && rm -fr go.sum && $(GOCMD) mod tidy
+
+.PHONY: fmt
+fmt: $(GOIMPORTS) $(GOFUMPT)
+	gofmt -w -s ./
+	$(GOIMPORTS) -w  ./
+	$(GOFUMPT) -l -w .
diff --git a/protobufs/anyvalue.pb.go b/protobufs/anyvalue.pb.go
index 697b805..453fd58 100644
--- a/protobufs/anyvalue.pb.go
+++ b/protobufs/anyvalue.pb.go
@@ -27,10 +27,11 @@
 package protobufs
 
 import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
 	sync "sync"
+
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 )
 
 const (
diff --git a/protobufs/opamp.pb.go b/protobufs/opamp.pb.go
index ce08bbd..e3fbddd 100644
--- a/protobufs/opamp.pb.go
+++ b/protobufs/opamp.pb.go
@@ -23,10 +23,11 @@
 package protobufs
 
 import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
 	sync "sync"
+
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 )
 
 const (
diff --git a/server/httpconnectioncontext.go b/server/httpconnectioncontext.go
index f35803a..6a90a17 100644
--- a/server/httpconnectioncontext.go
+++ b/server/httpconnectioncontext.go
@@ -17,6 +17,7 @@ func contextWithConn(ctx context.Context, c net.Conn) context.Context {
 	// of http.Server to remember the connection in the context.
 	return context.WithValue(ctx, connContextKey, c)
 }
+
 func connFromRequest(r *http.Request) net.Conn {
 	// Extract the net.Conn from the context of the specified http.Request.
 	return r.Context().Value(connContextKey).(net.Conn)
diff --git a/server/server.go b/server/server.go
index 46acff9..abcf538 100644
--- a/server/server.go
+++ b/server/server.go
@@ -73,8 +73,9 @@ type OpAMPServer interface {
 	// accept connections.
 	Start(settings StartSettings) error
 
-	// Stop accepting new connections and close all current connections. This should
-	// block until all connections are closed.
+	// Stop accepting new connections and close all current connections.
+	// This operation should block until both the server socket and all
+	// connections have been closed.
 	Stop(ctx context.Context) error
 
 	// Addr returns the network address Server is listening on. Nil if not started.
diff --git a/server/serverimpl.go b/server/serverimpl.go
index dcdd044..de9055d 100644
--- a/server/serverimpl.go
+++ b/server/serverimpl.go
@@ -5,6 +5,7 @@ import (
 	"compress/gzip"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -19,16 +20,16 @@ import (
 	serverTypes "github.com/open-telemetry/opamp-go/server/types"
 )
 
-var (
-	errAlreadyStarted = errors.New("already started")
-)
+var errAlreadyStarted = errors.New("already started")
 
-const defaultOpAMPPath = "/v1/opamp"
-const headerContentType = "Content-Type"
-const headerContentEncoding = "Content-Encoding"
-const headerAcceptEncoding = "Accept-Encoding"
-const contentEncodingGzip = "gzip"
-const contentTypeProtobuf = "application/x-protobuf"
+const (
+	defaultOpAMPPath      = "/v1/opamp"
+	headerContentType     = "Content-Type"
+	headerContentEncoding = "Content-Encoding"
+	headerAcceptEncoding  = "Accept-Encoding"
+	contentEncodingGzip   = "gzip"
+	contentTypeProtobuf   = "application/x-protobuf"
+)
 
 type server struct {
 	logger   types.Logger
@@ -39,7 +40,8 @@ type server struct {
 
 	// The listening HTTP Server after successful Start() call. Nil if Start()
 	// is not called or was not successful.
-	httpServer *http.Server
+	httpServer        *http.Server
+	httpServerServeWg *sync.WaitGroup
 
 	// The network address Server is listening on. Nil if not started.
 	addr net.Addr
@@ -108,6 +110,9 @@ func (s *server) Start(settings StartSettings) error {
 		ConnContext: contextWithConn,
 	}
 	s.httpServer = hs
+	httpServerServeWg := sync.WaitGroup{}
+	httpServerServeWg.Add(1)
+	s.httpServerServeWg = &httpServerServeWg
 
 	listenAddr := s.httpServer.Addr
 
@@ -118,7 +123,10 @@ func (s *server) Start(settings StartSettings) error {
 		}
 		err = s.startHttpServer(
 			listenAddr,
-			func(l net.Listener) error { return hs.ServeTLS(l, "", "") },
+			func(l net.Listener) error {
+				defer httpServerServeWg.Done()
+				return hs.ServeTLS(l, "", "")
+			},
 		)
 	} else {
 		if listenAddr == "" {
@@ -126,7 +134,10 @@ func (s *server) Start(settings StartSettings) error {
 		}
 		err = s.startHttpServer(
 			listenAddr,
-			func(l net.Listener) error { return hs.Serve(l) },
+			func(l net.Listener) error {
+				defer httpServerServeWg.Done()
+				return hs.Serve(l)
+			},
 		)
 	}
 	return err
@@ -159,7 +170,16 @@ func (s *server) Stop(ctx context.Context) error {
 		defer func() { s.httpServer = nil }()
 		// This stops accepting new connections. TODO: close existing
 		// connections and wait them to be terminated.
-		return s.httpServer.Shutdown(ctx)
+		err := s.httpServer.Shutdown(ctx)
+		if err != nil {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			s.httpServerServeWg.Wait()
+		}
 	}
 	return nil
 }
@@ -226,27 +246,39 @@ func (s *server) handleWSConnection(reqCtx context.Context, wsConn *websocket.Co
 	// Loop until fail to read from the WebSocket connection.
 	for {
 		msgContext := context.Background()
+		request := protobufs.AgentToServer{}
+
 		// Block until the next message can be read.
 		mt, msgBytes, err := wsConn.ReadMessage()
-		if err != nil {
-			if !websocket.IsUnexpectedCloseError(err) {
-				s.logger.Errorf(msgContext, "Cannot read a message from WebSocket: %v", err)
-				break
+		isBreak, err := func() (bool, error) {
+			if err != nil {
+				if !websocket.IsUnexpectedCloseError(err) {
+					s.logger.Errorf(msgContext, "Cannot read a message from WebSocket: %v", err)
+					return true, err
+				}
+				// This is a normal closing of the WebSocket connection.
+				s.logger.Debugf(msgContext, "Agent disconnected: %v", err)
+				return true, err
+			}
+			if mt != websocket.BinaryMessage {
+				err = fmt.Errorf("unexpected message type: %v, must be binary message", mt)
+				s.logger.Errorf(msgContext, "Cannot process a message from WebSocket: %v", err)
+				return false, err
 			}
-			// This is a normal closing of the WebSocket connection.
-			s.logger.Debugf(msgContext, "Agent disconnected: %v", err)
-			break
-		}
-		if mt != websocket.BinaryMessage {
-			s.logger.Errorf(msgContext, "Received unexpected message type from WebSocket: %v", mt)
-			continue
-		}
 
-		// Decode WebSocket message as a Protobuf message.
-		var request protobufs.AgentToServer
-		err = internal.DecodeWSMessage(msgBytes, &request)
+			// Decode WebSocket message as a Protobuf message.
+			err = internal.DecodeWSMessage(msgBytes, &request)
+			if err != nil {
+				s.logger.Errorf(msgContext, "Cannot decode message from WebSocket: %v", err)
+				return false, err
+			}
+			return false, nil
+		}()
 		if err != nil {
-			s.logger.Errorf(msgContext, "Cannot decode message from WebSocket: %v", err)
+			connectionCallbacks.OnReadMessageError(agentConn, mt, msgBytes, err)
+			if isBreak {
+				break
+			}
 			continue
 		}
 
@@ -366,7 +398,6 @@ func (s *server) handlePlainHTTPRequest(req *http.Request, w http.ResponseWriter
 		w.Header().Set(headerContentEncoding, contentEncodingGzip)
 	}
 	_, err = w.Write(bodyBytes)
-
 	if err != nil {
 		s.logger.Debugf(req.Context(), "Cannot send HTTP response: %v", err)
 	}
diff --git a/server/types/callbacks.go b/server/types/callbacks.go
index b25e818..9df3f21 100644
--- a/server/types/callbacks.go
+++ b/server/types/callbacks.go
@@ -62,6 +62,9 @@ type ConnectionCallbacks struct {
 
 	// OnConnectionClose is called when the OpAMP connection is closed.
 	OnConnectionClose func(conn Connection)
+
+	// OnReadMessageError is called when an error occurs while reading or deserializing a message.
+	OnReadMessageError func(conn Connection, mt int, msgByte []byte, err error)
 }
 
 func defaultOnConnected(ctx context.Context, conn Connection) {}
@@ -77,6 +80,8 @@ func defaultOnMessage(
 
 func defaultOnConnectionClose(conn Connection) {}
 
+func defaultOnReadMessageError(conn Connection, mt int, msgByte []byte, err error) {}
+
 func (c *ConnectionCallbacks) SetDefaults() {
 	if c.OnConnected == nil {
 		c.OnConnected = defaultOnConnected
@@ -89,4 +94,8 @@ func (c *ConnectionCallbacks) SetDefaults() {
 	if c.OnConnectionClose == nil {
 		c.OnConnectionClose = defaultOnConnectionClose
 	}
+
+	if c.OnReadMessageError == nil {
+		c.OnReadMessageError = defaultOnReadMessageError
+	}
 }


Test patch:
diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml
index 34a0944..80ef753 100644
--- a/.github/workflows/build-and-test.yml
+++ b/.github/workflows/build-and-test.yml
@@ -65,6 +65,10 @@ jobs:
         with:
           path: /home/runner/.cache/go-build
           key: go-build-unittest-${{ runner.os }}-${{ hashFiles('**/go.mod', '**/go.sum') }}
+      - name: Go fmt
+        run: |
+          make fmt
+          git diff --exit-code || (echo 'Code base is not Go formatted, please run "make fmt" and commit the changes in this PR.' && exit 1)
       - name: Build and Test
         run: make
 
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
index 87d3e5e..f16e278 100644
--- a/client/clientimpl_test.go
+++ b/client/clientimpl_test.go
@@ -468,7 +468,6 @@ func createRemoteConfig() *protobufs.AgentRemoteConfig {
 
 func TestFirstStatusReport(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		remoteConfig := createRemoteConfig()
 
 		// Start a Server.
@@ -629,13 +628,11 @@ func TestSetEffectiveConfig(t *testing.T) {
 		// Shutdown the client.
 		err := client.Stop(context.Background())
 		assert.NoError(t, err)
-
 	})
 }
 
 func TestSetAgentDescription(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		var rcvAgentDescr atomic.Value
@@ -927,7 +924,6 @@ func TestClientRequestConnectionSettings(t *testing.T) {
 
 func TestReportAgentDescription(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -990,7 +986,6 @@ func TestReportAgentDescription(t *testing.T) {
 
 func TestReportAgentHealth(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1063,7 +1058,6 @@ func TestReportAgentHealth(t *testing.T) {
 
 func TestReportEffectiveConfig(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1131,7 +1125,6 @@ func TestReportEffectiveConfig(t *testing.T) {
 
 func verifyRemoteConfigUpdate(t *testing.T, successCase bool, expectStatus *protobufs.RemoteConfigStatus) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1235,7 +1228,6 @@ func verifyRemoteConfigUpdate(t *testing.T, successCase bool, expectStatus *prot
 }
 
 func TestRemoteConfigUpdate(t *testing.T) {
-
 	tests := []struct {
 		name           string
 		success        bool
@@ -1280,7 +1272,8 @@ const packageUpdateErrorMsg = "cannot update packages"
 
 func assertPackageStatus(t *testing.T,
 	testCase packageTestCase,
-	msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool) {
+	msg *protobufs.AgentToServer,
+) (*protobufs.ServerToAgent, bool) {
 	expectedStatusReceived := false
 
 	status := msg.PackageStatuses
@@ -1327,7 +1320,6 @@ func assertPackageStatus(t *testing.T,
 
 func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1383,7 +1375,8 @@ func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 		// ---> Server
 		// Wait for the expected package statuses to be received.
 		srv.EventuallyExpect("full PackageStatuses", func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent,
-			bool) {
+			bool,
+		) {
 			return assertPackageStatus(t, testCase, msg)
 		})
 
@@ -1439,10 +1432,16 @@ const packageFileURL = "/validfile.pkg"
 
 var packageFileContent = []byte("Package File Content")
 
+var optionalAuthHeaders = protobufs.Header{Key: "Authorization", Value: "Basic YWxhZGRpbjpvcGVuc2VzYW1l"}
+
 func createDownloadSrv(t *testing.T) *httptest.Server {
 	m := http.NewServeMux()
 	m.HandleFunc(packageFileURL,
 		func(w http.ResponseWriter, r *http.Request) {
+			if r.Header.Get(optionalAuthHeaders.GetKey()) != "" {
+				assert.Equal(t, r.Header.Get(optionalAuthHeaders.GetKey()), optionalAuthHeaders.GetValue())
+			}
+
 			w.WriteHeader(http.StatusOK)
 			_, err := w.Write(packageFileContent)
 			assert.NoError(t, err)
@@ -1506,7 +1505,6 @@ func createPackageTestCase(name string, downloadSrv *httptest.Server) packageTes
 }
 
 func TestUpdatePackages(t *testing.T) {
-
 	downloadSrv := createDownloadSrv(t)
 	defer downloadSrv.Close()
 
@@ -1527,6 +1525,11 @@ func TestUpdatePackages(t *testing.T) {
 	errorOnCallback.errorOnCallback = true
 	tests = append(tests, errorOnCallback)
 
+	// A case where we send optional headers
+	withHeaders := createPackageTestCase("with optional HTTP headers", downloadSrv)
+	withHeaders.available.Packages["package1"].File.Headers = &protobufs.Headers{Headers: []*protobufs.Header{&optionalAuthHeaders}}
+	tests = append(tests, withHeaders)
+
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			verifyUpdatePackages(t, test)
@@ -1645,14 +1648,12 @@ func TestMissingPackagesStateProvider(t *testing.T) {
 }
 
 func TestOfferUpdatedVersion(t *testing.T) {
-
 	downloadSrv := createDownloadSrv(t)
 	defer downloadSrv.Close()
 
 	testCase := createPackageTestCase("offer new version", downloadSrv)
 
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		localPackageState := internal.NewInMemPackagesStore()
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1695,7 +1696,8 @@ func TestOfferUpdatedVersion(t *testing.T) {
 		// ---> Server
 		// Wait for the expected package statuses to be received.
 		srv.EventuallyExpect("full PackageStatuses", func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent,
-			bool) {
+			bool,
+		) {
 			return assertPackageStatus(t, testCase, msg)
 		})
 
@@ -1722,7 +1724,8 @@ func TestOfferUpdatedVersion(t *testing.T) {
 		// ---> Server
 		// Wait for the expected package statuses to be received.
 		srv.EventuallyExpect("full PackageStatuses updated version", func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent,
-			bool) {
+			bool,
+		) {
 			return assertPackageStatus(t, testCase, msg)
 		})
 
@@ -1737,7 +1740,6 @@ func TestOfferUpdatedVersion(t *testing.T) {
 
 func TestReportCustomCapabilities(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
@@ -1892,7 +1894,6 @@ func TestSendCustomMessage(t *testing.T) {
 // TestCustomMessages tests the custom messages functionality.
 func TestCustomMessages(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		var rcvCustomMessage atomic.Value
@@ -2125,7 +2126,6 @@ func TestCustomMessagesSendAndWait(t *testing.T) {
 // TestSetCustomCapabilities tests the ability for the client to change the set of custom capabilities that it supports.
 func TestSetCustomCapabilities(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		var rcvCustomCapabilities atomic.Value
@@ -2208,7 +2208,6 @@ func TestSetCustomCapabilities(t *testing.T) {
 // TestSetFlags tests the ability for the client to change the set of flags it sends.
 func TestSetFlags(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
-
 		// Start a Server.
 		srv := internal.StartMockServer(t)
 		var rcvCustomFlags atomic.Value
@@ -2361,7 +2360,6 @@ func TestSetAvailableComponents(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.desc, func(t *testing.T) {
 			testClients(t, func(t *testing.T, client OpAMPClient) {
-
 				// Start a Server.
 				srv := internal.StartMockServer(t)
 				srv.EnableExpectMode()
diff --git a/client/internal/httpsender_test.go b/client/internal/httpsender_test.go
index 78ff2f8..19eb1cc 100644
--- a/client/internal/httpsender_test.go
+++ b/client/internal/httpsender_test.go
@@ -21,7 +21,6 @@ import (
 )
 
 func TestHTTPSenderRetryForStatusTooManyRequests(t *testing.T) {
-
 	var connectionAttempts int64
 	srv := StartMockServer(t)
 	srv.OnRequest = func(w http.ResponseWriter, r *http.Request) {
@@ -98,7 +97,6 @@ func TestAddTLSConfig(t *testing.T) {
 }
 
 func GenerateCertificate() (tls.Certificate, error) {
-
 	certPem := []byte(`-----BEGIN CERTIFICATE-----
 MIIBhTCCASugAwIBAgIQIRi6zePL6mKjOipn+dNuaTAKBggqhkjOPQQDAjASMRAw
 DgYDVQQKEwdBY21lIENvMB4XDTE3MTAyMDE5NDMwNloXDTE4MTAyMDE5NDMwNlow
@@ -126,7 +124,6 @@ EKTcWGekdmdDPsHloRNtsiCa697B2O9IFA==
 }
 
 func TestHTTPSenderRetryForFailedRequests(t *testing.T) {
-
 	srv, m := newMockServer(t)
 	address := testhelpers.GetAvailableLocalAddress()
 	var connectionAttempts int64
diff --git a/client/internal/wsreceiver_test.go b/client/internal/wsreceiver_test.go
index c8ade42..0bbbf87 100644
--- a/client/internal/wsreceiver_test.go
+++ b/client/internal/wsreceiver_test.go
@@ -41,7 +41,6 @@ const (
 )
 
 func TestServerToAgentCommand(t *testing.T) {
-
 	tests := []struct {
 		command *protobufs.ServerToAgentCommand
 		action  commandAction
@@ -188,7 +187,6 @@ func TestDecodeMessage(t *testing.T) {
 }
 
 func TestReceiverLoopStop(t *testing.T) {
-
 	srv := StartMockServer(t)
 
 	conn, _, err := websocket.DefaultDialer.DialContext(
diff --git a/client/wsclient_test.go b/client/wsclient_test.go
index 66b89e0..4bba316 100644
--- a/client/wsclient_test.go
+++ b/client/wsclient_test.go
@@ -194,11 +194,9 @@ func TestDisconnectWSByServer(t *testing.T) {
 }
 
 func TestVerifyWSCompress(t *testing.T) {
-
 	tests := []bool{false, true}
 	for _, withCompression := range tests {
 		t.Run(fmt.Sprintf("%v", withCompression), func(t *testing.T) {
-
 			// Start a Server.
 			srv := internal.StartMockServer(t)
 			srv.EnableExpectMode()
diff --git a/internal/examples/supervisor/supervisor/supervisor_test.go b/internal/examples/supervisor/supervisor/supervisor_test.go
index 4568720..7239488 100644
--- a/internal/examples/supervisor/supervisor/supervisor_test.go
+++ b/internal/examples/supervisor/supervisor/supervisor_test.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"os"
 	"testing"
-	
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/open-telemetry/opamp-go/internal"
@@ -51,9 +51,9 @@ func TestNewSupervisor(t *testing.T) {
 server:
   endpoint: ws://127.0.0.1:4320/v1/opamp
 agent:
-  executable: %s/dummy_agent.sh`, tmpDir)), 0644)
+  executable: %s/dummy_agent.sh`, tmpDir)), 0o644)
 
-	os.WriteFile("dummy_agent.sh", []byte("#!/bin/sh\nsleep 9999\n"), 0755)
+	os.WriteFile("dummy_agent.sh", []byte("#!/bin/sh\nsleep 9999\n"), 0o755)
 
 	startOpampServer(t)
 
diff --git a/server/serverimpl_test.go b/server/serverimpl_test.go
index ca7a5c3..5fa95f9 100644
--- a/server/serverimpl_test.go
+++ b/server/serverimpl_test.go
@@ -57,6 +57,36 @@ func TestServerStartStop(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestServerStartStopWithCancel(t *testing.T) {
+	srv := startServer(t, &StartSettings{})
+
+	err := srv.Start(StartSettings{})
+	assert.ErrorIs(t, err, errAlreadyStarted)
+
+	canceledCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err = srv.Stop(canceledCtx)
+	assert.ErrorIs(t, err, context.Canceled)
+}
+
+func TestServerStartStopIdempotency(t *testing.T) {
+	endpoint := testhelpers.GetAvailableLocalAddress()
+	for i := 0; i < 10; i++ {
+		t.Run(fmt.Sprintf("Attempt #%d: ", i), func(t *testing.T) {
+			srv := startServer(t, &StartSettings{
+				ListenEndpoint: endpoint,
+			})
+
+			err := srv.Start(StartSettings{})
+			assert.ErrorIs(t, err, errAlreadyStarted)
+
+			err = srv.Stop(context.Background())
+			assert.NoError(t, err)
+		})
+	}
+}
+
 func TestServerStartStopWithMiddleware(t *testing.T) {
 	var addedMiddleware atomic.Bool
 	assert.False(t, addedMiddleware.Load())
@@ -315,6 +345,52 @@ func TestServerReceiveSendMessage(t *testing.T) {
 	assert.EqualValues(t, settings.CustomCapabilities, response.CustomCapabilities.Capabilities)
 }
 
+func TestServerReceiveSendErrorMessage(t *testing.T) {
+	var rcvMsg atomic.Value
+	type ErrorInfo struct {
+		mt      int
+		msgByte []byte
+		err     error
+	}
+	callbacks := types.Callbacks{
+		OnConnecting: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true, ConnectionCallbacks: types.ConnectionCallbacks{
+				OnReadMessageError: func(conn types.Connection, mt int, msgByte []byte, err error) {
+					rcvMsg.Store(ErrorInfo{
+						mt:      mt,
+						msgByte: msgByte,
+						err:     err,
+					})
+				},
+			}}
+		},
+	}
+
+	// Start a Server.
+	settings := &StartSettings{Settings: Settings{
+		Callbacks:          callbacks,
+		CustomCapabilities: []string{"local.test.capability"},
+	}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Connect using a WebSocket client.
+	conn, _, _ := dialClient(settings)
+	require.NotNil(t, conn)
+	defer conn.Close()
+
+	// Send an invalid message to the Server. This should result in calling OnReadMessageError().
+	err := conn.WriteMessage(websocket.TextMessage, []byte("abc"))
+	require.NoError(t, err)
+
+	// Wait until Server receives the message.
+	eventually(t, func() bool { return rcvMsg.Load() != nil })
+	errInfo := rcvMsg.Load().(ErrorInfo)
+	assert.EqualValues(t, websocket.TextMessage, errInfo.mt)
+	assert.EqualValues(t, []byte("abc"), errInfo.msgByte)
+	assert.NotNil(t, errInfo.err)
+}
+
 func TestServerReceiveSendMessageWithCompression(t *testing.T) {
 	// Use highly compressible config body.
 	uncompressedCfg := []byte(strings.Repeat("test", 10000))
@@ -620,7 +696,6 @@ func TestServerAttachSendMessagePlainHTTP(t *testing.T) {
 }
 
 func TestServerHonoursClientRequestContentEncoding(t *testing.T) {
-
 	hc := http.Client{}
 	var rcvMsg atomic.Value
 	var onConnectedCalled, onCloseCalled int32
@@ -698,7 +773,6 @@ func TestServerHonoursClientRequestContentEncoding(t *testing.T) {
 }
 
 func TestServerHonoursAcceptEncoding(t *testing.T) {
-
 	hc := http.Client{}
 	var rcvMsg atomic.Value
 	var onConnectedCalled, onCloseCalled int32
@@ -985,7 +1059,6 @@ func BenchmarkSendToClient(b *testing.B) {
 	}
 	srv := New(&sharedinternal.NopLogger{})
 	err := srv.Start(*settings)
-
 	if err != nil {
 		b.Error(err)
 	}
@@ -1017,7 +1090,6 @@ func BenchmarkSendToClient(b *testing.B) {
 
 	for _, conn := range serverConnections {
 		err := conn.Send(context.Background(), &protobufs.ServerToAgent{})
-
 		if err != nil {
 			b.Error(err)
 		}
@@ -1026,5 +1098,4 @@ func BenchmarkSendToClient(b *testing.B) {
 	for _, conn := range clientConnections {
 		conn.Close()
 	}
-
 }