Upload nginx config

internal/bus/topics.go

@@ -19,4 +19,5 @@ const (
 	InstanceHealthTopic              = "instance-health"
 	ConfigUploadRequestTopic         = "config-upload-request"
 	DataPlaneResponseTopic           = "data-plane-response"
+	ConnectionCreatedTopic           = "connection-created"
 )

internal/command/command_plugin.go

@@ -23,8 +23,9 @@ var _ bus.Plugin = (*CommandPlugin)(nil)
 
 type (
 	commandService interface {
-		UpdateDataPlaneStatus(ctx context.Context, resource *mpi.Resource) error
+		UpdateDataPlaneStatus(ctx context.Context, resource *mpi.Resource) (*mpi.CreateConnectionResponse, error)
 		UpdateDataPlaneHealth(ctx context.Context, instanceHealths []*mpi.InstanceHealth) error
+		SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error
 		CancelSubscription(ctx context.Context)
 	}
 
@@ -67,32 +68,54 @@ func (cp *CommandPlugin) Info() *bus.Info {
 	}
 }
 
-// nolint: revive
 func (cp *CommandPlugin) Process(ctx context.Context, msg *bus.Message) {
 	switch msg.Topic {
 	case bus.ResourceUpdateTopic:
-		if resource, ok := msg.Data.(*mpi.Resource); ok {
-			err := cp.commandService.UpdateDataPlaneStatus(ctx, resource)
-			if err != nil {
-				slog.ErrorContext(ctx, "Unable to update data plane status", "error", err)
-			}
-		}
+		cp.processResourceUpdate(ctx, msg)
 	case bus.InstanceHealthTopic:
-		if instances, ok := msg.Data.([]*mpi.InstanceHealth); ok {
-			err := cp.commandService.UpdateDataPlaneHealth(ctx, instances)
-			if err != nil {
-				slog.ErrorContext(ctx, "Unable to update data plane health", "error", err)
-			}
-		}
+		cp.processInstanceHealth(ctx, msg)
+	case bus.DataPlaneResponseTopic:
+		cp.processDataPlaneResponse(ctx, msg)
 	default:
 		slog.DebugContext(ctx, "Command plugin unknown topic", "topic", msg.Topic)
 	}
 }
 
+func (cp *CommandPlugin) processResourceUpdate(ctx context.Context, msg *bus.Message) {
+	if resource, ok := msg.Data.(*mpi.Resource); ok {
+		createConnectionResponse, err := cp.commandService.UpdateDataPlaneStatus(ctx, resource)
+		if err != nil {
+			slog.ErrorContext(ctx, "Unable to update data plane status", "error", err)
+		}
+		if createConnectionResponse != nil {
+			cp.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConnectionCreatedTopic, Data: createConnectionResponse})
+		}
+	}
+}
+
+func (cp *CommandPlugin) processInstanceHealth(ctx context.Context, msg *bus.Message) {
+	if instances, ok := msg.Data.([]*mpi.InstanceHealth); ok {
+		err := cp.commandService.UpdateDataPlaneHealth(ctx, instances)
+		if err != nil {
+			slog.ErrorContext(ctx, "Unable to update data plane health", "error", err)
+		}
+	}
+}
+
+func (cp *CommandPlugin) processDataPlaneResponse(ctx context.Context, msg *bus.Message) {
+	if response, ok := msg.Data.(*mpi.DataPlaneResponse); ok {
+		err := cp.commandService.SendDataPlaneResponse(ctx, response)
+		if err != nil {
+			slog.ErrorContext(ctx, "Unable to send data plane response", "error", err)
+		}
+	}
+}
+
 func (cp *CommandPlugin) Subscriptions() []string {
 	return []string{
 		bus.ResourceUpdateTopic,
 		bus.InstanceHealthTopic,
+		bus.DataPlaneResponseTopic,
 	}
 }
 
@@ -108,6 +131,13 @@ func (cp *CommandPlugin) monitorSubscribeChannel(ctx context.Context) {
 				slog.Any(logger.CorrelationIDKey, message.GetMessageMeta().GetCorrelationId()),
 			)
 			slog.DebugContext(newCtx, "Received management plane request", "request", message)
+
+			switch message.GetRequest().(type) {
+			case *mpi.ManagementPlaneRequest_ConfigUploadRequest:
+				cp.messagePipe.Process(ctx, &bus.Message{Topic: bus.ConfigUploadRequestTopic, Data: message})
+			default:
+				slog.DebugContext(newCtx, "Management plane request not implemented yet")
+			}
 		}
 	}
 }

internal/command/command_plugin_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
 	"github.com/nginx/agent/v3/internal/bus"
 	"github.com/nginx/agent/v3/internal/command/commandfakes"
 	"github.com/nginx/agent/v3/internal/grpc/grpcfakes"
@@ -38,6 +39,7 @@ func TestCommandPlugin_Subscriptions(t *testing.T) {
 		[]string{
 			bus.ResourceUpdateTopic,
 			bus.InstanceHealthTopic,
+			bus.DataPlaneResponseTopic,
 		},
 		subscriptions,
 	)
@@ -79,6 +81,44 @@ func TestCommandPlugin_Process(t *testing.T) {
 
 	commandPlugin.Process(ctx, &bus.Message{Topic: bus.InstanceHealthTopic, Data: protos.GetInstanceHealths()})
 	require.Equal(t, 1, fakeCommandService.UpdateDataPlaneHealthCallCount())
+
+	commandPlugin.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: protos.OKDataPlaneResponse()})
+	require.Equal(t, 1, fakeCommandService.SendDataPlaneResponseCallCount())
+}
+
+func TestCommandPlugin_monitorSubscribeChannel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	messagePipe := bus.NewFakeMessagePipe()
+
+	commandPlugin := NewCommandPlugin(types.AgentConfig(), &grpcfakes.FakeGrpcConnectionInterface{})
+	err := commandPlugin.Init(ctx, messagePipe)
+	require.NoError(t, err)
+	defer commandPlugin.Close(ctx)
+
+	go commandPlugin.monitorSubscribeChannel(ctx)
+
+	commandPlugin.subscribeChannel <- &mpi.ManagementPlaneRequest{
+		Request: &mpi.ManagementPlaneRequest_ConfigUploadRequest{
+			ConfigUploadRequest: &mpi.ConfigUploadRequest{},
+		},
+	}
+
+	assert.Eventually(
+		t,
+		func() bool { return len(messagePipe.GetMessages()) == 1 },
+		2*time.Second,
+		10*time.Millisecond,
+	)
+
+	messages := messagePipe.GetMessages()
+	assert.Len(t, messages, 1)
+	assert.Equal(t, bus.ConfigUploadRequestTopic, messages[0].Topic)
+
+	request, ok := messages[0].Data.(*mpi.ManagementPlaneRequest)
+	assert.True(t, ok)
+	require.NotNil(t, request.GetConfigUploadRequest())
 }
 
 func TestMonitorSubscribeChannel(t *testing.T) {