Repository: open-telemetry/opamp-go
Pull request #87 (2022-05-26 19:55:38)

Refactor OpAMPClient callbacks
Resolves https://github.com/open-telemetry/opamp-go/issues/77

We previously delivered the data received in one ServerToAgent message using
multiple callbacks. This encouraged piecemeal processing and resulted in
multiple state changes when only one would be sufficient.

The new OnMessage callback delivers all data that can affect the state at once
which makes it possible process it in one pass. In the Supervisor this allows
to restart the Agent only once when applying changes that result in config
file change.

codecov[bot]: # [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) Report
> Merging [#87](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (e49bb94) into [main](https://codecov.io/gh/open-telemetry/opamp-go/commit/0403a82c5871e6c2970d98d286d15df1d5f5c89d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (0403a82) will **decrease** coverage by `2.26%`.
> The diff coverage is `78.65%`.

```diff
@@            Coverage Diff             @@
##             main      #87      +/-   ##
==========================================
- Coverage   78.08%   75.81%   -2.27%     
==========================================
  Files          22       22              
  Lines        1720     1650      -70     
==========================================
- Hits         1343     1251      -92     
- Misses        283      299      +16     
- Partials       94      100       +6     
```


| [Impacted Files](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | Coverage Δ | |
|---|---|---|
| [client/internal/clientstate.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2NsaWVudHN0YXRlLmdv) | `81.57% <ø> (ø)` | |
| [client/internal/clientcommon.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2NsaWVudGNvbW1vbi5nbw==) | `82.23% <66.66%> (-3.49%)` | :arrow_down: |
| [client/internal/receivedprocessor.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL3JlY2VpdmVkcHJvY2Vzc29yLmdv) | `78.57% <83.33%> (-10.21%)` | :arrow_down: |
| [client/httpclient.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2h0dHBjbGllbnQuZ28=) | `94.23% <100.00%> (+0.48%)` | :arrow_up: |
| [client/types/callbacks.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L3R5cGVzL2NhbGxiYWNrcy5nbw==) | `66.66% <100.00%> (-5.00%)` | :arrow_down: |
| [client/wsclient.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L3dzY2xpZW50Lmdv) | `84.27% <100.00%> (+0.40%)` | :arrow_up: |
| [client/internal/httpsender.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/87/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2h0dHBzZW5kZXIuZ28=) | `72.26% <0.00%> (-9.49%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Last update [0403a82...e49bb94](https://codecov.io/gh/open-telemetry/opamp-go/pull/87?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).


Patch:
diff --git a/client/client.go b/client/client.go
index 5b0174c..5265f5d 100644
--- a/client/client.go
+++ b/client/client.go
@@ -47,6 +47,7 @@ type OpAMPClient interface {
 	// May be also called after Start(), in which case the attributes will be included
 	// in the next outgoing status report. This is typically used by Agents which allow
 	// their AgentDescription to change dynamically while the OpAMPClient is started.
+	// May be also called from OnMessage handler.
 	//
 	// The Hash field will be calculated and updated from the content of the rest of
 	// the fields.
@@ -59,5 +60,18 @@ type OpAMPClient interface {
 
 	// UpdateEffectiveConfig fetches the current local effective config using
 	// GetEffectiveConfig callback and sends it to the Server.
+	// May be called anytime after Start(), including from OnMessage handler.
 	UpdateEffectiveConfig(ctx context.Context) error
+
+	// SetRemoteConfigStatus sets the current RemoteConfigStatus.
+	// LastRemoteConfigHash field must be non-nil.
+	// May be called anytime after Start(), including from OnMessage handler.
+	// nil values are not allowed and will return an error.
+	SetRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error
+
+	// SetPackageStatuses sets the current PackageStatuses.
+	// ServerProvidedAllPackagesHash must be non-nil.
+	// May be called anytime after Start(), including from OnMessage handler.
+	// nil values are not allowed and will return an error.
+	SetPackageStatuses(statuses *protobufs.PackageStatuses) error
 }
diff --git a/client/httpclient.go b/client/httpclient.go
index a74b0f9..a18f2e0 100644
--- a/client/httpclient.go
+++ b/client/httpclient.go
@@ -72,6 +72,14 @@ func (c *httpClient) UpdateEffectiveConfig(ctx context.Context) error {
 	return c.common.UpdateEffectiveConfig(ctx)
 }
 
+func (c *httpClient) SetRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error {
+	return c.common.SetRemoteConfigStatus(status)
+}
+
+func (c *httpClient) SetPackageStatuses(statuses *protobufs.PackageStatuses) error {
+	return c.common.SetPackageStatuses(statuses)
+}
+
 func (c *httpClient) runUntilStopped(ctx context.Context) {
 	// Start the HTTP sender. This will make request/responses with retries for
 	// failures and will wait with configured polling interval if there is nothing
diff --git a/client/internal/clientcommon.go b/client/internal/clientcommon.go
index 6dbf234..84e52fb 100644
--- a/client/internal/clientcommon.go
+++ b/client/internal/clientcommon.go
@@ -1,6 +1,7 @@
 package internal
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"errors"
@@ -51,10 +52,14 @@ type ClientCommon struct {
 }
 
 func NewClientCommon(logger types.Logger, sender Sender) ClientCommon {
-	return ClientCommon{Logger: logger, sender: sender, stoppedSignal: make(chan struct{}, 1)}
+	return ClientCommon{
+		Logger: logger, sender: sender, stoppedSignal: make(chan struct{}, 1),
+	}
 }
 
-func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSettings) error {
+func (c *ClientCommon) PrepareStart(
+	_ context.Context, settings types.StartSettings,
+) error {
 	if c.isStarted {
 		return errAlreadyStarted
 	}
@@ -205,9 +210,11 @@ func (c *ClientCommon) SetAgentDescription(descr *protobufs.AgentDescription) er
 	if err := c.ClientSyncedState.SetAgentDescription(descr); err != nil {
 		return err
 	}
-	c.sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
-		msg.AgentDescription = c.ClientSyncedState.AgentDescription()
-	})
+	c.sender.NextMessage().Update(
+		func(msg *protobufs.AgentToServer) {
+			msg.AgentDescription = c.ClientSyncedState.AgentDescription()
+		},
+	)
 	c.sender.ScheduleSend()
 	return nil
 }
@@ -254,9 +261,13 @@ func (c *ClientCommon) UpdateEffectiveConfig(ctx context.Context) error {
 		calcHashEffectiveConfig(cfg)
 	}
 	// Send it to the Server.
-	c.sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
-		msg.EffectiveConfig = cfg
-	})
+	c.sender.NextMessage().Update(
+		func(msg *protobufs.AgentToServer) {
+			msg.EffectiveConfig = cfg
+		},
+	)
+	// TODO: if this call is coming from OnMessage callback don't schedule the send
+	// immediately, wait until the end of OnMessage to send one message only.
 	c.sender.ScheduleSend()
 
 	// Note that we do not store the EffectiveConfig anywhere else. It will be deleted
@@ -264,3 +275,61 @@ func (c *ClientCommon) UpdateEffectiveConfig(ctx context.Context) error {
 	// in memory for longer than it is needed.
 	return nil
 }
+
+func (c *ClientCommon) SetRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error {
+	if status.LastRemoteConfigHash == nil {
+		return errLastRemoteConfigHashNil
+	}
+
+	// Get the hash of the status before we update it.
+	prevHash := c.ClientSyncedState.RemoteConfigStatus().GetHash()
+
+	// Remember the new status.
+	if err := c.ClientSyncedState.SetRemoteConfigStatus(status); err != nil {
+		return err
+	}
+
+	// Check if the new status is different from the previous by comparing the hashes.
+	if !bytes.Equal(prevHash, status.Hash) {
+		// Let the Server know about the new status.
+		c.sender.NextMessage().Update(
+			func(msg *protobufs.AgentToServer) {
+				msg.RemoteConfigStatus = c.ClientSyncedState.RemoteConfigStatus()
+			},
+		)
+		// TODO: if this call is coming from OnMessage callback don't schedule the send
+		// immediately, wait until the end of OnMessage to send one message only.
+		c.sender.ScheduleSend()
+	}
+
+	return nil
+}
+
+func (c *ClientCommon) SetPackageStatuses(statuses *protobufs.PackageStatuses) error {
+	if statuses.ServerProvidedAllPackagesHash == nil {
+		return errServerProvidedAllPackagesHashNil
+	}
+
+	// Get the hash of the status before we update it.
+	prevHash := c.ClientSyncedState.PackageStatuses().GetHash()
+
+	if err := c.ClientSyncedState.SetPackageStatuses(statuses); err != nil {
+		return err
+	}
+
+	// Check if the new status is different from the previous by comparing the hashes.
+	if !bytes.Equal(prevHash, statuses.Hash) {
+		// Let the Server know about the new status.
+
+		c.sender.NextMessage().Update(
+			func(msg *protobufs.AgentToServer) {
+				msg.PackageStatuses = c.ClientSyncedState.PackageStatuses()
+			},
+		)
+		// TODO: if this call is coming from OnMessage callback don't schedule the send
+		// immediately, wait until the end of OnMessage to send one message only.
+		c.sender.ScheduleSend()
+	}
+
+	return nil
+}
diff --git a/client/internal/clientstate.go b/client/internal/clientstate.go
index be74d56..33b5f31 100644
--- a/client/internal/clientstate.go
+++ b/client/internal/clientstate.go
@@ -13,8 +13,10 @@ import (
 )
 
 var (
-	errRemoteConfigStatusMissing = errors.New("RemoteConfigStatus is not set")
-	errPackageStatusesMissing    = errors.New("PackageStatuses is not set")
+	errRemoteConfigStatusMissing        = errors.New("RemoteConfigStatus is not set")
+	errLastRemoteConfigHashNil          = errors.New("LastRemoteConfigHash is nil")
+	errPackageStatusesMissing           = errors.New("PackageStatuses is not set")
+	errServerProvidedAllPackagesHashNil = errors.New("ServerProvidedAllPackagesHash is nil")
 )
 
 // ClientSyncedState stores the state of the Agent messages that the OpAMP Client needs to
diff --git a/client/internal/receivedprocessor.go b/client/internal/receivedprocessor.go
index fedaaf2..1c1e27b 100644
--- a/client/internal/receivedprocessor.go
+++ b/client/internal/receivedprocessor.go
@@ -1,8 +1,8 @@
 package internal
 
 import (
-	"bytes"
 	"context"
+	"errors"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -52,18 +52,43 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 			return
 		}
 
-		scheduled := r.rcvRemoteConfig(ctx, msg.RemoteConfig)
-
-		r.rcvConnectionSettings(ctx, msg.ConnectionSettings)
-		scheduled = scheduled || r.rcvPackagesAvailable(ctx, msg.PackagesAvailable)
-		r.rcvAgentIdentification(ctx, msg.AgentIdentification)
-
-		scheduled2, err := r.rcvFlags(ctx, msg.Flags)
+		scheduled, err := r.rcvFlags(ctx, msg.Flags)
 		if err != nil {
 			r.logger.Errorf("cannot processed received flags:%v", err)
 		}
 
-		scheduled = scheduled || scheduled2
+		msgData := &types.MessageData{
+			RemoteConfig: msg.RemoteConfig,
+		}
+
+		if msg.ConnectionSettings != nil {
+			msgData.OwnMetricsConnSettings = msg.ConnectionSettings.OwnMetrics
+			msgData.OwnTracesConnSettings = msg.ConnectionSettings.OwnTraces
+			msgData.OwnLogsConnSettings = msg.ConnectionSettings.OwnLogs
+			msgData.OtherConnSettings = msg.ConnectionSettings.OtherConnections
+		}
+
+		if msg.PackagesAvailable != nil {
+			msgData.PackagesAvailable = msg.PackagesAvailable
+			msgData.PackageSyncer = NewPackagesSyncer(
+				r.logger,
+				msgData.PackagesAvailable,
+				r.sender,
+				r.clientSyncedState,
+				r.packagesStateProvider,
+			)
+		}
+
+		if msg.AgentIdentification != nil {
+			err := r.rcvAgentIdentification(msg.AgentIdentification)
+			if err == nil {
+				msgData.AgentIdentification = msg.AgentIdentification
+			}
+		}
+
+		r.callbacks.OnMessage(ctx, msgData)
+
+		r.rcvOpampConnectionSettings(ctx, msg.ConnectionSettings)
 
 		if scheduled {
 			r.sender.ScheduleSend()
@@ -121,124 +146,15 @@ func (r *receivedProcessor) rcvFlags(
 	return scheduleSend, nil
 }
 
-func (r *receivedProcessor) rcvRemoteConfig(
-	ctx context.Context,
-	remoteCfg *protobufs.AgentRemoteConfig,
-) (reportStatus bool) {
-	if remoteCfg == nil {
-		return false
-	}
-
-	// Ask the Agent to apply the remote config.
-	effective, changed, applyErr := r.callbacks.OnRemoteConfig(ctx, remoteCfg)
-
-	// Begin creating the new status.
-	cfgStatus := &protobufs.RemoteConfigStatus{}
-
-	if applyErr != nil {
-		// Remote config applying failed.
-		cfgStatus.Status = protobufs.RemoteConfigStatus_FAILED
-		cfgStatus.ErrorMessage = applyErr.Error()
-	} else {
-		// Applied successfully.
-		cfgStatus.Status = protobufs.RemoteConfigStatus_APPLIED
-	}
-
-	// Respond back with the hash of the config received from the Server.
-	cfgStatus.LastRemoteConfigHash = remoteCfg.ConfigHash
-
-	// Get the hash of the status before we update it.
-	prevStatus := r.clientSyncedState.RemoteConfigStatus()
-	var prevCfgStatusHash []byte
-	if prevStatus != nil {
-		prevCfgStatusHash = prevStatus.Hash
-	}
-
-	// Remember the status for the future use.
-	_ = r.clientSyncedState.SetRemoteConfigStatus(cfgStatus)
-
-	// Check if the status has changed by comparing the hashes.
-	if !bytes.Equal(prevCfgStatusHash, cfgStatus.Hash) {
-		// Let the Agent know about the status.
-		r.callbacks.SaveRemoteConfigStatus(ctx, cfgStatus)
-
-		// Include the config status in the next message to the Server.
-		r.sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
-			msg.RemoteConfigStatus = cfgStatus
-		})
-
-		reportStatus = true
-	}
-
-	// Report the effective configuration if it changed and remote config was applied
-	// successfully.
-	if changed && applyErr == nil {
-		r.sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
-			msg.EffectiveConfig = effective
-		})
-		reportStatus = true
-	}
-	return reportStatus
-}
-
-func (r *receivedProcessor) rcvConnectionSettings(ctx context.Context, settings *protobufs.ConnectionSettingsOffers) {
-	if settings == nil {
+func (r *receivedProcessor) rcvOpampConnectionSettings(ctx context.Context, settings *protobufs.ConnectionSettingsOffers) {
+	if settings == nil || settings.Opamp == 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 *receivedProcessor) rcvOwnTelemetryConnectionSettings(
-	ctx context.Context,
-	settings *protobufs.TelemetryConnectionSettings,
-	telemetryType types.OwnTelemetryType,
-	callback func(ctx context.Context, telemetryType types.OwnTelemetryType, settings *protobufs.TelemetryConnectionSettings) error,
-) {
-	if settings != nil {
-		callback(ctx, telemetryType, settings)
-	}
-}
-
-func (r *receivedProcessor) rcvOtherConnectionSettings(
-	ctx context.Context,
-	settings *protobufs.OtherConnectionSettings,
-	name string,
-	callback func(ctx context.Context, name string, settings *protobufs.OtherConnectionSettings) error,
-) {
-	if settings != nil {
-		callback(ctx, name, settings)
+	err := r.callbacks.OnOpampConnectionSettings(ctx, settings.Opamp)
+	if err == nil {
+		// TODO: verify connection using new settings.
+		r.callbacks.OnOpampConnectionSettingsAccepted(settings.Opamp)
 	}
 }
 
@@ -247,65 +163,20 @@ func (r *receivedProcessor) processErrorResponse(body *protobufs.ServerErrorResp
 	r.logger.Errorf("received an error from server: %s", body.ErrorMessage)
 }
 
-func (r *receivedProcessor) rcvPackagesAvailable(ctx context.Context, available *protobufs.PackagesAvailable) bool {
-	if available == nil {
-		return false
-	}
-
-	syncer := NewPackagesSyncer(r.logger, available, r.sender, r.clientSyncedState, r.packagesStateProvider)
-
-	// The OnPackagesAvailable is expected to call syncer.Sync() which will
-	// do the syncing process in background.
-	err := r.callbacks.OnPackagesAvailable(ctx, available, syncer)
-	if err != nil {
-		// Could not even start syncing. Just report everything failed.
-		statuses := &protobufs.PackageStatuses{
-			ServerProvidedAllPackagesHash: available.AllPackagesHash,
-			Packages:                      map[string]*protobufs.PackageStatus{},
-			// TODO: we need a way to report general errors that occurred when trying
-			// install available packages, errors which are not related to a particular
-			// package.
-		}
-		// Until the TODO above is done we will set the same error message for all
-		// packages.
-		for name, pkg := range available.Packages {
-			statuses.Packages[name] = &protobufs.PackageStatus{
-				Name:                 name,
-				ServerOfferedVersion: pkg.Version,
-				ServerOfferedHash:    pkg.Hash,
-				Status:               protobufs.PackageStatus_InstallFailed,
-				ErrorMessage:         err.Error(),
-			}
-		}
-		_ = r.clientSyncedState.SetPackageStatuses(statuses)
-		r.sender.NextMessage().Update(
-			func(msg *protobufs.AgentToServer) { msg.PackageStatuses = statuses },
-		)
-	}
-
-	return true
-}
-
-func (r *receivedProcessor) rcvAgentIdentification(ctx context.Context, agentId *protobufs.AgentIdentification) {
-	if agentId == nil {
-		return
-	}
-
+func (r *receivedProcessor) rcvAgentIdentification(agentId *protobufs.AgentIdentification) error {
 	if agentId.NewInstanceUid == "" {
-		r.logger.Debugf("Empty instance uid is not allowed")
-		return
-	}
-
-	err := r.callbacks.OnAgentIdentification(ctx, agentId)
-	if err != nil {
-		r.logger.Errorf("Error while updating agent identification: %v", err)
-		return
+		err := errors.New("empty instance uid is not allowed")
+		r.logger.Debugf(err.Error())
+		return err
 	}
 
-	err = r.sender.SetInstanceUid(agentId.NewInstanceUid)
+	err := r.sender.SetInstanceUid(agentId.NewInstanceUid)
 	if err != nil {
 		r.logger.Errorf("Error while setting instance uid: %v, err")
+		return err
 	}
+
+	return nil
 }
 
 func (r *receivedProcessor) rcvCommand(command *protobufs.ServerToAgentCommand) {
diff --git a/client/types/callbacks.go b/client/types/callbacks.go
index 3e1303e..f8090c5 100644
--- a/client/types/callbacks.go
+++ b/client/types/callbacks.go
@@ -6,13 +6,37 @@ import (
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-type OwnTelemetryType int
-
-const (
-	OwnMetrics OwnTelemetryType = iota
-	OwnTraces
-	OwnLogs
-)
+type MessageData struct {
+	// RemoteConfig is offered by the Server. The Agent must process it and call
+	// OpAMPClient.SetRemoteConfigStatus to indicate success or failure. If the
+	// effective config has changed as a result of processing the Agent must also call
+	// OpAMPClient.UpdateEffectiveConfig. SetRemoteConfigStatus and UpdateEffectiveConfig
+	// may be called from OnMessage handler or after OnMessage returns.
+	RemoteConfig *protobufs.AgentRemoteConfig
+
+	// Connection settings are offered by the Server. These fields should be processed
+	// as described in the ConnectionSettingsOffers message.
+	OwnMetricsConnSettings *protobufs.TelemetryConnectionSettings
+	OwnTracesConnSettings  *protobufs.TelemetryConnectionSettings
+	OwnLogsConnSettings    *protobufs.TelemetryConnectionSettings
+	OtherConnSettings      map[string]*protobufs.OtherConnectionSettings
+
+	// PackagesAvailable offered by the Server. The Agent must process the offer.
+	// The typical way to process is to call PackageSyncer.Sync() function, which will
+	// take care of reporting the status to the Server as processing happens.
+	//
+	// If PackageSyncer.Sync() function is not called then it is the responsibility of
+	// OnMessage handler to do the processing and call OpAMPClient.SetPackageStatuses to
+	// reflect the processing status. SetPackageStatuses may be called from OnMessage
+	// handler or after OnMessage returns.
+	PackagesAvailable *protobufs.PackagesAvailable
+	PackageSyncer     PackagesSyncer
+
+	// AgentIdentification indicates a new identification received from the Server.
+	// The Agent must save this identification and use it in the future instantiations
+	// of OpAMPClient.
+	AgentIdentification *protobufs.AgentIdentification
+}
 
 type Callbacks interface {
 	OnConnect()
@@ -28,45 +52,15 @@ type Callbacks interface {
 	// ErrorResponse_UNAVAILABLE case internally by performing retries as necessary.
 	OnError(err *protobufs.ServerErrorResponse)
 
-	// OnRemoteConfig is called when the Agent receives a remote config from the Server.
-	// The config parameter will not be nil.
-	//
-	// The Agent should process the config and return the effective config if processing
-	// succeeded or an error if processing failed.
-	//
-	// configChanged must be set to true if as a result of applying the remote config
-	// the effective config has changed.
-	//
-	// The returned effective config or the error will be reported back to the Server
-	// via AgentToServer message (using EffectiveConfig and RemoteConfigStatus fields).
-	//
-	// Only one OnRemoteConfig call can be active at any time. Until OnRemoteConfig
-	// returns it will not be called again. Any other remote configs received from
-	// the Server while OnRemoteConfig call has not returned will be remembered.
-	// Once OnRemoteConfig call returns it will be called again with the most recent
-	// remote config received.
-	//
-	// The EffectiveConfig.Hash field will be calculated and updated from the content
-	// of the rest of the fields.
-	OnRemoteConfig(
-		ctx context.Context,
-		remoteConfig *protobufs.AgentRemoteConfig,
-	) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error)
-
-	// SaveRemoteConfigStatus is called after OnRemoteConfig returns. The status
-	// will be set either as APPLIED or FAILED depending on whether OnRemoteConfig
-	// returned a success or error.
-	// The Agent must remember this RemoteConfigStatus and supply in the future
-	// calls to Start() in StartSettings.RemoteConfigStatus.
-	SaveRemoteConfigStatus(ctx context.Context, status *protobufs.RemoteConfigStatus)
-
-	// GetEffectiveConfig returns the current effective config. Only one
-	// GetEffectiveConfig call can be active at any time. Until GetEffectiveConfig
-	// returns it will not be called again.
-	//
-	// The Hash field in the returned EffectiveConfig will be calculated and updated
-	// by the caller from the content of the rest of the fields.
-	GetEffectiveConfig(ctx context.Context) (*protobufs.EffectiveConfig, error)
+	// OnMessage is called when the Agent receives a message that needs processing.
+	// See MessageData definition for the data that may be available for processing.
+	// During OnMessage execution the OpAMPClient functions that change the status
+	// of the client may be called, e.g. if RemoteConfig is processed then
+	// SetRemoteConfigStatus should be called to reflect the processing result.
+	// These functions may also be called after OnMessage returns. This is advisable
+	// if processing can take a long time. In that case returning quickly is preferable
+	// to avoid blocking the OpAMPClient.
+	OnMessage(ctx context.Context, msg *MessageData)
 
 	// OnOpampConnectionSettings is called when the Agent receives an OpAMP
 	// connection settings offer from the Server. Typically the settings can specify
@@ -75,8 +69,7 @@ type Callbacks interface {
 	//
 	// The Agent should process the offer and return an error if the Agent does not
 	// want to accept the settings (e.g. if the TSL certificate in the settings
-	// cannot be verified). The returned error will be reported back to the Server
-	// via AgentToServer message (using ConnectionStatuses field).
+	// cannot be verified).
 	//
 	// If OnOpampConnectionSettings returns nil and then the caller will
 	// attempt to reconnect to the OpAMP Server using the new settings.
@@ -84,9 +77,6 @@ type Callbacks interface {
 	// be reported to the Server. If the connection succeeds the new settings
 	// will be used by the client from that moment on.
 	//
-	// Accepted or rejected settings will be reported back to the Server via
-	// ConnectionStatuses message.
-	//
 	// Only one OnOpampConnectionSettings call can be active at any time.
 	// See OnRemoteConfig for the behavior.
 	OnOpampConnectionSettings(
@@ -102,55 +92,27 @@ type Callbacks interface {
 		settings *protobufs.OpAMPConnectionSettings,
 	)
 
-	// OnOwnTelemetryConnectionSettings is called when the Agent receives a
-	// connection settings to be used for reporting Agent's own telemetry.
-	//
-	// The Agent should process the settings and return an error if the Agent does not
-	// want to accept the settings (e.g. if the TSL certificate in the settings
-	// cannot be verified). The returned error will be reported back to the Server
-	// via AgentToServer message (using ConnectionStatuses field).
-	// If the Agent accepts the settings it should return nil and begin sending
-	// its own telemetry to the destination specified in the settings.
-	// We currently support 3 types of Agent's own telemetry: metrics, traces, logs.
-	// The Agent can support any subset of these types.
-	OnOwnTelemetryConnectionSettings(
-		ctx context.Context,
-		telemetryType OwnTelemetryType,
-		settings *protobufs.TelemetryConnectionSettings,
-	) error
-
-	// OnOtherConnectionSettings is called when the Agent receives a
-	// connection settings to be used for by the Agent for any other purposes.
-	// Typically these are used by the Agent to send collected data to the destinations
-	// it is configured to. The name is typically the name of the destination.
-	//
-	// The Agent should process the settings and return an error if the Agent does not
-	// want to accept the settings (e.g. if the TSL certificate in the settings
-	// cannot be verified). The returned error will be reported back to the Server
-	// via AgentToServer message (using ConnectionStatuses field).
-	OnOtherConnectionSettings(
-		ctx context.Context,
-		name string,
-		certificate *protobufs.OtherConnectionSettings,
-	) error
+	// For all methods that accept a context parameter the caller may cancel the
+	// context if processing takes too long. In that case the method should return
+	// as soon as possible with an error.
 
-	// OnPackagesAvailable is called when the Server has packages available which are
-	// different from what the Agent indicated it has via
-	// LastServerProvidedAllPackagesHash.
-	// syncer can be used to initiate syncing the packages from the Server.
-	OnPackagesAvailable(ctx context.Context, packages *protobufs.PackagesAvailable, syncer PackagesSyncer) error
+	// SaveRemoteConfigStatus is called after OnRemoteConfig returns. The status
+	// will be set either as APPLIED or FAILED depending on whether OnRemoteConfig
+	// returned a success or error.
+	// The Agent must remember this RemoteConfigStatus and supply in the future
+	// calls to Start() in StartSettings.RemoteConfigStatus.
+	SaveRemoteConfigStatus(ctx context.Context, status *protobufs.RemoteConfigStatus)
 
-	// OnAgentIdentification is called when the Server requests changing identification of the Agent.
-	// Agent should be updated with new id and use it for all further communication.
-	// If Agent does not support the identification override from the Server, the function should return an error.
-	OnAgentIdentification(ctx context.Context, agentId *protobufs.AgentIdentification) error
+	// GetEffectiveConfig returns the current effective config. Only one
+	// GetEffectiveConfig call can be active at any time. Until GetEffectiveConfig
+	// returns it will not be called again.
+	//
+	// The Hash field in the returned EffectiveConfig will be calculated and updated
+	// by the caller from the content of the rest of the fields.
+	GetEffectiveConfig(ctx context.Context) (*protobufs.EffectiveConfig, error)
 
 	// OnCommand is called when the Server requests that the connected Agent perform a command.
 	OnCommand(command *protobufs.ServerToAgentCommand) error
-
-	// For all methods that accept a context parameter the caller may cancel the
-	// context if processing takes too long. In that case the method should return
-	// as soon as possible with an error.
 }
 
 type CallbacksStruct struct {
@@ -158,14 +120,7 @@ type CallbacksStruct struct {
 	OnConnectFailedFunc func(err error)
 	OnErrorFunc         func(err *protobufs.ServerErrorResponse)
 
-	OnRemoteConfigFunc func(
-		ctx context.Context,
-		remoteConfig *protobufs.AgentRemoteConfig,
-	) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error)
-
-	SaveRemoteConfigStatusFunc func(ctx context.Context, status *protobufs.RemoteConfigStatus)
-
-	GetEffectiveConfigFunc func(ctx context.Context) (*protobufs.EffectiveConfig, error)
+	OnMessageFunc func(ctx context.Context, msg *MessageData)
 
 	OnOpampConnectionSettingsFunc func(
 		ctx context.Context,
@@ -175,22 +130,10 @@ type CallbacksStruct struct {
 		settings *protobufs.OpAMPConnectionSettings,
 	)
 
-	OnOwnTelemetryConnectionSettingsFunc func(
-		ctx context.Context,
-		telemetryType OwnTelemetryType,
-		settings *protobufs.TelemetryConnectionSettings,
-	) error
-
-	OnOtherConnectionSettingsFunc func(
-		ctx context.Context,
-		name string,
-		settings *protobufs.OtherConnectionSettings,
-	) error
-
-	OnPackagesAvailableFunc   func(ctx context.Context, packages *protobufs.PackagesAvailable, syncer PackagesSyncer) error
-	OnAgentIdentificationFunc func(ctx context.Context, agentId *protobufs.AgentIdentification) error
-
 	OnCommandFunc func(command *protobufs.ServerToAgentCommand) error
+
+	SaveRemoteConfigStatusFunc func(ctx context.Context, status *protobufs.RemoteConfigStatus)
+	GetEffectiveConfigFunc     func(ctx context.Context) (*protobufs.EffectiveConfig, error)
 }
 
 var _ Callbacks = (*CallbacksStruct)(nil)
@@ -213,14 +156,10 @@ func (c CallbacksStruct) OnError(err *protobufs.ServerErrorResponse) {
 	}
 }
 
-func (c CallbacksStruct) OnRemoteConfig(
-	ctx context.Context,
-	remoteConfig *protobufs.AgentRemoteConfig,
-) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-	if c.OnRemoteConfigFunc != nil {
-		return c.OnRemoteConfigFunc(ctx, remoteConfig)
+func (c CallbacksStruct) OnMessage(ctx context.Context, msg *MessageData) {
+	if c.OnMessageFunc != nil {
+		c.OnMessageFunc(ctx, msg)
 	}
-	return nil, false, nil
 }
 
 func (c CallbacksStruct) SaveRemoteConfigStatus(ctx context.Context, status *protobufs.RemoteConfigStatus) {
@@ -251,49 +190,9 @@ func (c CallbacksStruct) OnOpampConnectionSettingsAccepted(settings *protobufs.O
 	}
 }
 
-func (c CallbacksStruct) OnOwnTelemetryConnectionSettings(
-	ctx context.Context, telemetryType OwnTelemetryType,
-	settings *protobufs.TelemetryConnectionSettings,
-) error {
-	if c.OnOwnTelemetryConnectionSettingsFunc != nil {
-		return c.OnOwnTelemetryConnectionSettingsFunc(ctx, telemetryType, settings)
-	}
-	return nil
-}
-
-func (c CallbacksStruct) OnOtherConnectionSettings(
-	ctx context.Context, name string, settings *protobufs.OtherConnectionSettings,
-) error {
-	if c.OnOtherConnectionSettingsFunc != nil {
-		return c.OnOtherConnectionSettingsFunc(ctx, name, settings)
-	}
-	return nil
-}
-
-func (c CallbacksStruct) OnPackagesAvailable(
-	ctx context.Context,
-	packages *protobufs.PackagesAvailable,
-	syncer PackagesSyncer,
-) error {
-	if c.OnPackagesAvailableFunc != nil {
-		return c.OnPackagesAvailableFunc(ctx, packages, syncer)
-	}
-	return nil
-}
-
 func (c CallbacksStruct) OnCommand(command *protobufs.ServerToAgentCommand) error {
 	if c.OnCommandFunc != nil {
 		return c.OnCommandFunc(command)
 	}
 	return nil
 }
-
-func (c CallbacksStruct) OnAgentIdentification(
-	ctx context.Context,
-	agentId *protobufs.AgentIdentification,
-) error {
-	if c.OnAgentIdentificationFunc != nil {
-		return c.OnAgentIdentificationFunc(ctx, agentId)
-	}
-	return nil
-}
diff --git a/client/wsclient.go b/client/wsclient.go
index 6d028b0..c0f9e34 100644
--- a/client/wsclient.go
+++ b/client/wsclient.go
@@ -101,6 +101,14 @@ func (c *wsClient) UpdateEffectiveConfig(ctx context.Context) error {
 	return c.common.UpdateEffectiveConfig(ctx)
 }
 
+func (c *wsClient) SetRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error {
+	return c.common.SetRemoteConfigStatus(status)
+}
+
+func (c *wsClient) SetPackageStatuses(statuses *protobufs.PackageStatuses) error {
+	return c.common.SetPackageStatuses(statuses)
+}
+
 // 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.
diff --git a/internal/examples/agent/agent/agent.go b/internal/examples/agent/agent/agent.go
index fb0e9b5..cf1b27f 100644
--- a/internal/examples/agent/agent/agent.go
+++ b/internal/examples/agent/agent/agent.go
@@ -96,15 +96,13 @@ func (agent *Agent) start() error {
 			OnErrorFunc: func(err *protobufs.ServerErrorResponse) {
 				agent.logger.Errorf("Server returned an error response: %v", err.ErrorMessage)
 			},
-			OnRemoteConfigFunc: agent.onRemoteConfig,
 			SaveRemoteConfigStatusFunc: func(_ context.Context, status *protobufs.RemoteConfigStatus) {
 				agent.remoteConfigStatus = status
 			},
-			OnOwnTelemetryConnectionSettingsFunc: agent.onOwnTelemetryConnectionSettings,
-			OnAgentIdentificationFunc:            agent.onAgentIdentificationFunc,
 			GetEffectiveConfigFunc: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
 				return agent.composeEffectiveConfig(), nil
 			},
+			OnMessageFunc: agent.onMessage,
 		},
 		RemoteConfigStatus: agent.remoteConfigStatus,
 	}
@@ -206,43 +204,6 @@ func (agent *Agent) composeEffectiveConfig() *protobufs.EffectiveConfig {
 	}
 }
 
-func (agent *Agent) onRemoteConfig(
-	_ context.Context,
-	remoteConfig *protobufs.AgentRemoteConfig,
-) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-	configChanged, err = agent.applyRemoteConfig(remoteConfig)
-	if err != nil {
-		return nil, false, err
-	}
-
-	return agent.composeEffectiveConfig(), configChanged, nil
-}
-
-func (agent *Agent) onOwnTelemetryConnectionSettings(
-	_ context.Context,
-	telemetryType types.OwnTelemetryType,
-	settings *protobufs.TelemetryConnectionSettings,
-) error {
-	switch telemetryType {
-	case types.OwnMetrics:
-		agent.initMeter(settings)
-	}
-
-	return nil
-}
-
-func (agent *Agent) onAgentIdentificationFunc(
-	_ context.Context,
-	agentId *protobufs.AgentIdentification,
-) error {
-	newInstanceId, err := ulid.Parse(agentId.NewInstanceUid)
-	if err != nil {
-		return err
-	}
-	agent.updateAgentIdentity(newInstanceId)
-	return nil
-}
-
 func (agent *Agent) initMeter(settings *protobufs.TelemetryConnectionSettings) {
 	reporter, err := NewMetricReporter(agent.logger, settings, agent.agentType, agent.agentVersion, agent.instanceId)
 	if err != nil {
@@ -357,3 +318,42 @@ func (agent *Agent) Shutdown() {
 		_ = agent.opampClient.Stop(context.Background())
 	}
 }
+
+func (agent *Agent) onMessage(ctx context.Context, msg *types.MessageData) {
+	configChanged := false
+	if msg.RemoteConfig != nil {
+		var err error
+		configChanged, err = agent.applyRemoteConfig(msg.RemoteConfig)
+		if err != nil {
+			agent.opampClient.SetRemoteConfigStatus(&protobufs.RemoteConfigStatus{
+				LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+				Status:               protobufs.RemoteConfigStatus_FAILED,
+				ErrorMessage:         err.Error(),
+			})
+		} else {
+			agent.opampClient.SetRemoteConfigStatus(&protobufs.RemoteConfigStatus{
+				LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+				Status:               protobufs.RemoteConfigStatus_APPLIED,
+			})
+		}
+	}
+
+	if msg.OwnMetricsConnSettings != nil {
+		agent.initMeter(msg.OwnMetricsConnSettings)
+	}
+
+	if msg.AgentIdentification != nil {
+		newInstanceId, err := ulid.Parse(msg.AgentIdentification.NewInstanceUid)
+		if err != nil {
+			agent.logger.Errorf(err.Error())
+		}
+		agent.updateAgentIdentity(newInstanceId)
+	}
+
+	if configChanged {
+		err := agent.opampClient.UpdateEffectiveConfig(ctx)
+		if err != nil {
+			agent.logger.Errorf(err.Error())
+		}
+	}
+}
diff --git a/internal/examples/supervisor/supervisor/supervisor.go b/internal/examples/supervisor/supervisor/supervisor.go
index cba2540..99147d0 100644
--- a/internal/examples/supervisor/supervisor/supervisor.go
+++ b/internal/examples/supervisor/supervisor/supervisor.go
@@ -145,12 +145,10 @@ func (s *Supervisor) startOpAMP() error {
 			OnErrorFunc: func(err *protobufs.ServerErrorResponse) {
 				s.logger.Errorf("Server returned an error response: %v", err.ErrorMessage)
 			},
-			OnRemoteConfigFunc:                   s.onRemoteConfig,
-			OnOwnTelemetryConnectionSettingsFunc: s.onOwnTelemetryConnectionSettings,
-			OnAgentIdentificationFunc:            s.onAgentIdentificationFunc,
 			GetEffectiveConfigFunc: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
 				return s.createEffectiveConfigMsg(), nil
 			},
+			OnMessageFunc: s.onMessage,
 		},
 	}
 	err := s.opampClient.SetAgentDescription(s.createAgentDescription())
@@ -243,67 +241,15 @@ func (s *Supervisor) createEffectiveConfigMsg() *protobufs.EffectiveConfig {
 	return cfg
 }
 
-// onRemoteConfig processes a remote config message received from OpAMP Server.
-func (s *Supervisor) onRemoteConfig(
-	_ context.Context,
-	remoteConfig *protobufs.AgentRemoteConfig,
-) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-	if remoteConfig == nil {
-		return nil, false, nil
-	}
-
-	s.remoteConfig = remoteConfig
-	s.logger.Debugf("Received remote config from server, hash=%x.", s.remoteConfig.ConfigHash)
-
-	configChanged, err = s.recalcEffectiveConfig()
-	if err != nil {
-		return nil, false, err
-	}
-
-	return s.createEffectiveConfigMsg(), configChanged, nil
-}
-
-// onOwnTelemetryConnectionSettings processes a own telemetry settings message
-// received from OpAMP Server.
-func (s *Supervisor) onOwnTelemetryConnectionSettings(
-	ctx context.Context,
-	telemetryType types.OwnTelemetryType,
-	settings *protobufs.TelemetryConnectionSettings,
-) error {
-	switch telemetryType {
-	case types.OwnMetrics:
-		s.setupOwnMetrics(ctx, settings)
-	}
-
-	return nil
-}
-
-// onAgentIdentificationFunc processes new identification message
-// received from OpAMP Server.
-func (s *Supervisor) onAgentIdentificationFunc(
-	_ context.Context,
-	agentId *protobufs.AgentIdentification,
-) error {
-	newInstanceId, err := ulid.Parse(agentId.NewInstanceUid)
-	if err != nil {
-		return err
-	}
-
-	s.logger.Debugf("Agent identify is being changed from id=%v to id=%v",
-		s.instanceId.String(),
-		newInstanceId.String())
-	s.instanceId = newInstanceId
-
-	// TODO: update metrics pipeline.
-	return nil
-}
-
-func (s *Supervisor) setupOwnMetrics(ctx context.Context, settings *protobufs.TelemetryConnectionSettings) {
+func (s *Supervisor) setupOwnMetrics(ctx context.Context, settings *protobufs.TelemetryConnectionSettings) (configChanged bool) {
 	var cfg string
 	if settings.DestinationEndpoint == "" {
 		// No destination. Disable metric collection.
+		s.logger.Debugf("Disabling own metrics pipeline in the config")
 		cfg = ""
 	} else {
+		s.logger.Debugf("Enabling own metrics pipeline in the config")
+
 		// TODO: choose the scraping port dynamically instead of hard-coding to 8888.
 		cfg = fmt.Sprintf(
 			`
@@ -337,13 +283,7 @@ service:
 		return
 	}
 
-	if configChanged {
-		// TODO: make explicit in the OpAMPClient contract that we are allowed to call
-		// UpdateEffectiveConfig from the callbacks (like it is done here).
-		// This currently works, but is not ideal since it may result in 2 messages sent:
-		// one as a result of this call and another when the callback returns.
-		s.opampClient.UpdateEffectiveConfig(ctx)
-	}
+	return configChanged
 }
 
 // composeEffectiveConfig composes the effective config from multiple sources:
@@ -427,21 +367,6 @@ func (s *Supervisor) recalcEffectiveConfig() (configChanged bool, err error) {
 		return configChanged, err
 	}
 
-	// TODO: delay the application of this config in case there is another callback coming
-	// that will change the connection settings and will result in effective config
-	// change again. Currently we unnecessarily restart the agent twice.
-
-	if configChanged {
-		s.logger.Debugf("Config is changed. Signal to restart the agent.")
-		// Signal that there is a new config.
-		select {
-		case s.hasNewConfig <- struct{}{}:
-		default:
-		}
-	} else {
-		s.logger.Debugf("Config is unchanged, nothing to do.")
-	}
-
 	return configChanged, nil
 }
 
@@ -508,3 +433,59 @@ func (s *Supervisor) Shutdown() {
 		_ = s.opampClient.Stop(context.Background())
 	}
 }
+
+func (s *Supervisor) onMessage(ctx context.Context, msg *types.MessageData) {
+	configChanged := false
+	if msg.RemoteConfig != nil {
+		s.remoteConfig = msg.RemoteConfig
+		s.logger.Debugf("Received remote config from server, hash=%x.", s.remoteConfig.ConfigHash)
+
+		var err error
+		configChanged, err = s.recalcEffectiveConfig()
+		if err != nil {
+			s.opampClient.SetRemoteConfigStatus(&protobufs.RemoteConfigStatus{
+				LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+				Status:               protobufs.RemoteConfigStatus_FAILED,
+				ErrorMessage:         err.Error(),
+			})
+		} else {
+			s.opampClient.SetRemoteConfigStatus(&protobufs.RemoteConfigStatus{
+				LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+				Status:               protobufs.RemoteConfigStatus_APPLIED,
+			})
+		}
+	}
+
+	if msg.OwnMetricsConnSettings != nil {
+		configChanged = s.setupOwnMetrics(ctx, msg.OwnMetricsConnSettings) || configChanged
+	}
+
+	if msg.AgentIdentification != nil {
+		newInstanceId, err := ulid.Parse(msg.AgentIdentification.NewInstanceUid)
+		if err != nil {
+			s.logger.Errorf(err.Error())
+		}
+
+		s.logger.Debugf("Agent identify is being changed from id=%v to id=%v",
+			s.instanceId.String(),
+			newInstanceId.String())
+		s.instanceId = newInstanceId
+
+		// TODO: update metrics pipeline by altering configuration and setting
+		// the instance id when Collector implements https://github.com/open-telemetry/opentelemetry-collector/pull/5402.
+	}
+
+	if configChanged {
+		err := s.opampClient.UpdateEffectiveConfig(ctx)
+		if err != nil {
+			s.logger.Errorf(err.Error())
+		}
+
+		s.logger.Debugf("Config is changed. Signal to restart the agent.")
+		// Signal that there is a new config.
+		select {
+		case s.hasNewConfig <- struct{}{}:
+		default:
+		}
+	}
+}
diff --git a/internal/proto/opamp.proto b/internal/proto/opamp.proto
index 8aa6e4c..80bac53 100644
--- a/internal/proto/opamp.proto
+++ b/internal/proto/opamp.proto
@@ -307,9 +307,7 @@ message TLSCertificate {
 
 message ConnectionSettingsOffers {
     // Hash of all settings, including settings that may be omitted from this message
-    // because they are unchanged. The Agent should remember the hash and include
-    // it in the subsequent ConnectionStatuses message, in the last_connection_settings_hash
-    // field.
+    // because they are unchanged.
     bytes hash = 1;
 
     // Settings to connect to the OpAMP Server.
diff --git a/protobufs/opamp.pb.go b/protobufs/opamp.pb.go
index e5625e6..e3ff2bf 100644
--- a/protobufs/opamp.pb.go
+++ b/protobufs/opamp.pb.go
@@ -1334,9 +1334,7 @@ type ConnectionSettingsOffers struct {
 	unknownFields protoimpl.UnknownFields
 
 	// Hash of all settings, including settings that may be omitted from this message
-	// because they are unchanged. The Agent should remember the hash and include
-	// it in the subsequent ConnectionStatuses message, in the last_connection_settings_hash
-	// field.
+	// because they are unchanged.
 	Hash []byte `protobuf:"bytes,1,opt,name=hash,proto3" json:"hash,omitempty"`
 	// Settings to connect to the OpAMP Server.
 	// If this field is not set then the Agent should assume that the settings are


Test patch:
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
index b3637ae..7327593 100644
--- a/client/clientimpl_test.go
+++ b/client/clientimpl_test.go
@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"errors"
 	"math/rand"
 	"net/http"
 	"net/http/httptest"
@@ -325,20 +324,11 @@ func TestFirstStatusReport(t *testing.T) {
 				OnConnectFunc: func() {
 					atomic.AddInt64(&connected, 1)
 				},
-				OnRemoteConfigFunc: func(
-					ctx context.Context,
-					config *protobufs.AgentRemoteConfig,
-				) (
-					effectiveConfig *protobufs.EffectiveConfig, configChanged bool,
-					err error,
-				) {
+				OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
 					// Verify that the client received exactly the remote config that
 					// the Server sent.
-					assert.True(t, proto.Equal(remoteConfig, config))
+					assert.True(t, proto.Equal(remoteConfig, msg.RemoteConfig))
 					atomic.AddInt64(&remoteConfigReceived, 1)
-					return &protobufs.EffectiveConfig{
-						ConfigMap: remoteConfig.Config,
-					}, true, nil
 				},
 			},
 		}
@@ -381,12 +371,6 @@ func TestIncludesDetailsOnReconnect(t *testing.T) {
 			OnConnectFunc: func() {
 				atomic.AddInt64(&connected, 1)
 			},
-			OnRemoteConfigFunc: func(
-				ctx context.Context,
-				config *protobufs.AgentRemoteConfig,
-			) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-				return &protobufs.EffectiveConfig{}, false, nil
-			},
 		},
 	}
 
@@ -565,14 +549,6 @@ func TestAgentIdentification(t *testing.T) {
 		// Start a client.
 		settings := types.StartSettings{}
 		settings.OpAMPServerURL = "ws://" + srv.Endpoint
-		settings.Callbacks = types.CallbacksStruct{
-			OnAgentIdentificationFunc: func(
-				ctx context.Context,
-				agentId *protobufs.AgentIdentification,
-			) error {
-				return nil
-			},
-		}
 		prepareClient(t, &settings, client)
 
 		oldInstanceUid := settings.InstanceUid
@@ -654,6 +630,17 @@ func TestConnectionSettings(t *testing.T) {
 		// Start a client.
 		settings := types.StartSettings{
 			Callbacks: types.CallbacksStruct{
+				OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
+					assert.True(t, proto.Equal(metricsSettings, msg.OwnMetricsConnSettings))
+					assert.True(t, proto.Equal(tracesSettings, msg.OwnTracesConnSettings))
+					assert.True(t, proto.Equal(logsSettings, msg.OwnLogsConnSettings))
+					atomic.AddInt64(&gotOwnSettings, 1)
+
+					assert.Len(t, msg.OtherConnSettings, 1)
+					assert.True(t, proto.Equal(otherSettings, msg.OtherConnSettings["other"]))
+					atomic.AddInt64(&gotOtherSettings, 1)
+				},
+
 				OnOpampConnectionSettingsFunc: func(
 					ctx context.Context, settings *protobufs.OpAMPConnectionSettings,
 				) error {
@@ -661,32 +648,6 @@ func TestConnectionSettings(t *testing.T) {
 					atomic.AddInt64(&gotOpampSettings, 1)
 					return nil
 				},
-
-				OnOwnTelemetryConnectionSettingsFunc: func(
-					ctx context.Context, telemetryType types.OwnTelemetryType,
-					settings *protobufs.TelemetryConnectionSettings,
-				) 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.OtherConnectionSettings,
-				) error {
-					assert.EqualValues(t, "other", name)
-					assert.True(t, proto.Equal(otherSettings, settings))
-					atomic.AddInt64(&gotOtherSettings, 1)
-					return nil
-				},
 			},
 		}
 		settings.OpAMPServerURL = "ws://" + srv.Endpoint
@@ -695,7 +656,7 @@ func TestConnectionSettings(t *testing.T) {
 		assert.NoError(t, client.Start(context.Background(), 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(&gotOwnSettings) == 1 })
 		eventually(t, func() bool { return atomic.LoadInt64(&gotOtherSettings) == 1 })
 		eventually(t, func() bool { return atomic.LoadInt64(&rcvStatus) == 1 })
 
@@ -848,22 +809,28 @@ func verifyRemoteConfigUpdate(t *testing.T, successCase bool, expectStatus *prot
 		srv := internal.StartMockServer(t)
 		srv.EnableExpectMode()
 
-		// Prepare a callback that returns either success or failure.
-		onRemoteConfigFunc := func(
-			ctx context.Context, remoteConfig *protobufs.AgentRemoteConfig,
-		) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-			if successCase {
-				return createEffectiveConfig(), true, nil
-			} else {
-				return nil, false, errors.New("cannot update remote config")
-			}
-		}
-
 		// Start a client.
 		settings := types.StartSettings{
 			OpAMPServerURL: "ws://" + srv.Endpoint,
 			Callbacks: types.CallbacksStruct{
-				OnRemoteConfigFunc: onRemoteConfigFunc,
+				OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
+					if msg.RemoteConfig != nil {
+						if successCase {
+							client.SetRemoteConfigStatus(
+								&protobufs.RemoteConfigStatus{
+									LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+									Status:               protobufs.RemoteConfigStatus_APPLIED,
+								})
+						} else {
+							client.SetRemoteConfigStatus(
+								&protobufs.RemoteConfigStatus{
+									LastRemoteConfigHash: msg.RemoteConfig.ConfigHash,
+									Status:               protobufs.RemoteConfigStatus_FAILED,
+									ErrorMessage:         "cannot update remote config",
+								})
+						}
+					}
+				},
 			},
 		}
 		prepareClient(t, &settings, client)
@@ -909,6 +876,7 @@ func verifyRemoteConfigUpdate(t *testing.T, successCase bool, expectStatus *prot
 			// This time all fields except Hash must be unset. This is expected
 			// as compression in OpAMP.
 			status := msg.RemoteConfigStatus
+			require.NotNil(t, status)
 			assert.EqualValues(t, firstConfigStatus.Hash, status.Hash)
 			assert.EqualValues(t, protobufs.RemoteConfigStatus_UNSET, status.Status)
 			assert.EqualValues(t, "", status.ErrorMessage)
@@ -977,6 +945,7 @@ type packageTestCase struct {
 	available           *protobufs.PackagesAvailable
 	expectedStatus      *protobufs.PackageStatuses
 	expectedFileContent map[string][]byte
+	expectedError       string
 }
 
 const packageUpdateErrorMsg = "cannot update packages"
@@ -993,14 +962,18 @@ func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 		var syncerDoneCh <-chan struct{}
 
 		// Prepare a callback that returns either success or failure.
-		onPackagesAvailable := func(ctx context.Context, packages *protobufs.PackagesAvailable, syncer types.PackagesSyncer) error {
-			if testCase.errorOnCallback {
-				return errors.New(packageUpdateErrorMsg)
-			} else {
-				syncerDoneCh = syncer.Done()
-				err := syncer.Sync(ctx)
-				require.NoError(t, err)
-				return nil
+		onMessageFunc := func(ctx context.Context, msg *types.MessageData) {
+			if msg.PackageSyncer != nil {
+				if testCase.errorOnCallback {
+					client.SetPackageStatuses(&protobufs.PackageStatuses{
+						ServerProvidedAllPackagesHash: msg.PackagesAvailable.AllPackagesHash,
+						ErrorMessage:                  packageUpdateErrorMsg,
+					})
+				} else {
+					syncerDoneCh = msg.PackageSyncer.Done()
+					err := msg.PackageSyncer.Sync(ctx)
+					require.NoError(t, err)
+				}
 			}
 		}
 
@@ -1008,7 +981,7 @@ func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 		settings := types.StartSettings{
 			OpAMPServerURL: "ws://" + srv.Endpoint,
 			Callbacks: types.CallbacksStruct{
-				OnPackagesAvailableFunc: onPackagesAvailable,
+				OnMessageFunc: onMessageFunc,
 			},
 			PackagesStateProvider: localPackageState,
 		}
@@ -1042,6 +1015,11 @@ func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
 				assert.EqualValues(t, testCase.expectedStatus.ServerProvidedAllPackagesHash, status.ServerProvidedAllPackagesHash)
 				lastStatusHash = status.Hash
 
+				if testCase.expectedError != "" {
+					assert.EqualValues(t, testCase.expectedError, status.ErrorMessage)
+					return &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}, true
+				}
+
 				// Verify individual package statuses.
 				for name, pkgExpected := range testCase.expectedStatus.Packages {
 					pkgStatus := status.Packages[name]
@@ -1209,8 +1187,7 @@ func TestUpdatePackages(t *testing.T) {
 
 	// A case when OnPackagesAvailable callback returns an error.
 	errorOnCallback := createPackageTestCase("error on callback", downloadSrv)
-	errorOnCallback.expectedStatus.Packages["package1"].Status = protobufs.PackageStatus_InstallFailed
-	errorOnCallback.expectedStatus.Packages["package1"].ErrorMessage = packageUpdateErrorMsg
+	errorOnCallback.expectedError = packageUpdateErrorMsg
 	errorOnCallback.errorOnCallback = true
 	tests = append(tests, errorOnCallback)
 
diff --git a/client/internal/wsreceiver_test.go b/client/internal/wsreceiver_test.go
index 725eaed..c767469 100644
--- a/client/internal/wsreceiver_test.go
+++ b/client/internal/wsreceiver_test.go
@@ -2,7 +2,6 @@ package internal
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"testing"
 
@@ -86,16 +85,15 @@ func TestServerToAgentCommand(t *testing.T) {
 
 func TestServerToAgentCommandExclusive(t *testing.T) {
 	calledCommand := false
-	calledRemoteConfig := false
+	calledOnMessageConfig := false
 
 	callbacks := types.CallbacksStruct{
 		OnCommandFunc: func(command *protobufs.ServerToAgentCommand) error {
 			calledCommand = true
 			return nil
 		},
-		OnRemoteConfigFunc: func(ctx context.Context, remoteConfig *protobufs.AgentRemoteConfig) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-			calledRemoteConfig = true
-			return nil, false, nil
+		OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
+			calledOnMessageConfig = true
 		},
 	}
 	clientSyncedState := ClientSyncedState{}
@@ -107,30 +105,5 @@ func TestServerToAgentCommandExclusive(t *testing.T) {
 		RemoteConfig: &protobufs.AgentRemoteConfig{},
 	})
 	assert.Equal(t, true, calledCommand, "OnCommand should be called when a Command is specified")
-	assert.Equal(t, false, calledRemoteConfig, "OnRemoteConfig should not be called when a Command is specified")
-}
-
-func TestOnRemoteConfigReportsError(t *testing.T) {
-	expectMsg := "error occurred while handling configuration"
-	expectStatus := protobufs.RemoteConfigStatus_FAILED
-
-	callbacks := types.CallbacksStruct{
-		OnRemoteConfigFunc: func(ctx context.Context, remoteConfig *protobufs.AgentRemoteConfig) (effectiveConfig *protobufs.EffectiveConfig, configChanged bool, err error) {
-			return nil, false, errors.New(expectMsg)
-		},
-	}
-
-	sender := NewSender(TestLogger{t})
-	clientSyncedState := ClientSyncedState{}
-	receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, sender, &clientSyncedState, nil)
-
-	receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
-		RemoteConfig: &protobufs.AgentRemoteConfig{},
-	})
-
-	gotStatus := receiver.sender.NextMessage().nextMessage.RemoteConfigStatus.Status
-	gotMsg := receiver.sender.NextMessage().nextMessage.RemoteConfigStatus.ErrorMessage
-
-	assert.Equal(t, expectStatus, gotStatus)
-	assert.Equal(t, expectMsg, gotMsg)
+	assert.Equal(t, false, calledOnMessageConfig, "OnMessage should not be called when a Command is specified")
 }