Allow OpAMP server to request an agent restart or shutdown

client/clientimpl_test.go

@@ -67,7 +67,7 @@ func TestConnectNoServer(t *testing.T) {
 func TestOnConnectFail(t *testing.T) {
 	var connectErr atomic.Value
 	settings := createNoServerSettings()
-	settings.Callbacks = CallbacksStruct{
+	settings.Callbacks = types.CallbacksStruct{
 		OnConnectFailedFunc: func(err error) {
 			connectErr.Store(err)
 		},
@@ -117,7 +117,7 @@ func TestConnectWithServer(t *testing.T) {
 	// Start a client.
 	var connected int64
 	settings := StartSettings{
-		Callbacks: CallbacksStruct{
+		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.StoreInt64(&connected, 1)
 			},
@@ -154,7 +154,7 @@ func TestConnectWithServer503(t *testing.T) {
 	var clientConnected int64
 	var connectErr atomic.Value
 	settings := StartSettings{
-		Callbacks: CallbacksStruct{
+		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.StoreInt64(&clientConnected, 1)
 				assert.Fail(t, "Client should not be able to connect")
@@ -217,7 +217,7 @@ func TestDisconnectByServer(t *testing.T) {
 	var connected int64
 	var connectErr atomic.Value
 	settings := StartSettings{
-		Callbacks: CallbacksStruct{
+		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.StoreInt64(&connected, 1)
 			},
@@ -266,7 +266,7 @@ func TestFirstStatusReport(t *testing.T) {
 	// Start a client.
 	var connected, remoteConfigReceived int64
 	settings := StartSettings{
-		Callbacks: CallbacksStruct{
+		Callbacks: types.CallbacksStruct{
 			OnConnectFunc: func() {
 				atomic.AddInt64(&connected, 1)
 			},
@@ -472,7 +472,7 @@ func TestConnectionSettings(t *testing.T) {
 
 	// Start a client.
 	settings := StartSettings{
-		Callbacks: CallbacksStruct{
+		Callbacks: types.CallbacksStruct{
 			OnOpampConnectionSettingsFunc: func(
 				ctx context.Context, settings *protobufs.ConnectionSettings,
 			) error {

client/internal/receiver.go

@@ -61,6 +61,12 @@ func (r *Receiver) receiveMessage(msg *protobufs.ServerToAgent) error {
 
 func (r *Receiver) processReceivedMessage(ctx context.Context, msg *protobufs.ServerToAgent) {
 	if r.callbacks != nil {
+		// If a command message exists, other messages will be ignored
+		if msg.Command != nil {
+			r.rcvCommand(msg.Command)
+			return
+		}
+
 		reportStatus := r.rcvRemoteConfig(ctx, msg.RemoteConfig, msg.Flags)
 
 		r.rcvConnectionSettings(ctx, msg.ConnectionSettings)
@@ -168,3 +174,9 @@ func (r *Receiver) processErrorResponse(body *protobufs.ServerErrorResponse) {
 func (r *Receiver) rcvAddonsAvailable(addons *protobufs.AddonsAvailable) {
 	// TODO: implement this.
 }
+
+func (r *Receiver) rcvCommand(command *protobufs.ServerToAgentCommand) {
+	if command != nil {
+		r.callbacks.OnCommand(command)
+	}
+}

client/internal/receiver_test.go

@@ -0,0 +1,105 @@
+package internal
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+type TestLogger struct {
+	*testing.T
+}
+
+func (logger TestLogger) Debugf(format string, v ...interface{}) {
+	logger.Logf(format, v...)
+}
+
+type commandAction int
+
+const (
+	none commandAction = iota
+	restart
+	unknown
+)
+
+func TestServerToAgentCommand(t *testing.T) {
+
+	tests := []struct {
+		command *protobufs.ServerToAgentCommand
+		action  commandAction
+		message string
+	}{
+		{
+			command: nil,
+			action:  none,
+			message: "No command should result in no action",
+		},
+		{
+			command: &protobufs.ServerToAgentCommand{
+				Type: protobufs.ServerToAgentCommand_Restart,
+			},
+			action:  restart,
+			message: "A Restart command should result in a restart",
+		},
+		{
+			command: &protobufs.ServerToAgentCommand{
+				Type: -1,
+			},
+			action:  unknown,
+			message: "An unknown command is still passed to the OnCommand callback",
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprint(i), func(t *testing.T) {
+			action := none
+
+			callbacks := types.CallbacksStruct{
+				OnCommandFunc: func(command *protobufs.ServerToAgentCommand) error {
+					switch command.Type {
+					case protobufs.ServerToAgentCommand_Restart:
+						action = restart
+					default:
+						action = unknown
+					}
+					return nil
+				},
+			}
+			receiver := NewReceiver(TestLogger{t}, callbacks, nil, nil)
+			receiver.processReceivedMessage(context.Background(), &protobufs.ServerToAgent{
+				Command: test.command,
+			})
+			assert.Equal(t, test.action, action, test.message)
+		})
+	}
+}
+
+func TestServerToAgentCommandExclusive(t *testing.T) {
+	calledCommand := false
+	calledRemoteConfig := 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
+		},
+	}
+	receiver := NewReceiver(TestLogger{t}, callbacks, nil, nil)
+	receiver.processReceivedMessage(context.Background(), &protobufs.ServerToAgent{
+		Command: &protobufs.ServerToAgentCommand{
+			Type: protobufs.ServerToAgentCommand_Restart,
+		},
+		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")
+}