bldr/plugin/plugin-server-resource_test.go

Rama master · réplica de lectura del repositorio original.

package bldr_plugin

import (
	"context"
	"testing"

	"github.com/aperturerobotics/controllerbus/controller"
	"github.com/aperturerobotics/controllerbus/core"
	"github.com/aperturerobotics/starpc/rpcstream"
	"github.com/aperturerobotics/starpc/srpc"
	"github.com/pkg/errors"
	"github.com/s4wave/spacewave/bldr/resource"
	bifrost_rpc "github.com/s4wave/spacewave/net/rpc"
	"github.com/sirupsen/logrus"
)

func TestPluginServerResourceServiceFallsThroughToBus(t *testing.T) {
	// Build a core bus to host the resource service invoker.
	ctx := t.Context()

	// Attach a logger for bus diagnostics.
	log := logrus.New()
	le := logrus.NewEntry(log)
	b, _, err := core.NewCoreBus(ctx, le)
	if err != nil {
		t.Fatal(err)
	}

	// Register a sentinel resource service on the target mux.
	targetMux := srpc.NewMux()
	if err := resource.SRPCRegisterResourceService(targetMux, sentinelResourceService{}); err != nil {
		t.Fatal(err)
	}
	ctrl := bifrost_rpc.NewInvokerController(
		le,
		b,
		controller.NewInfo("bldr/plugin/test-resource-service", controller.MustParseVersion("0.0.1"), ""),
		targetMux,
		[]string{resource.SRPCResourceServiceServiceID},
	)
	rel, err := b.AddController(ctx, ctrl, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer rel()

	// Reach the sentinel service through the plugin server's RPC stream.
	pluginMux := srpc.NewMux()
	if err := SRPCRegisterPlugin(pluginMux, NewPluginServer(b)); err != nil {
		t.Fatal(err)
	}
	pluginClient := NewSRPCPluginClient(srpc.NewClient(srpc.NewServerPipe(srpc.NewServer(pluginMux))))
	openStream := rpcstream.NewRpcStreamClient(pluginClient.PluginRpc, "spacewave-app", true)
	resClient := resource.NewSRPCResourceServiceClient(openStream)

	// Open a resource client stream and complete the init handshake.
	strm, err := resClient.ResourceClient(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer strm.Close()
	if err := strm.Send(&resource.ResourceClientRequest{
		Body: &resource.ResourceClientRequest_Init{
			Init: &resource.ResourceClientInitRequest{},
		},
	}); err != nil {
		t.Fatal(err)
	}
	resp, err := strm.Recv()
	if err != nil {
		t.Fatal(err)
	}

	// The sentinel service answered, so the plugin server fell through to the bus.
	init := resp.GetInit()
	if init == nil {
		t.Fatalf("expected init response, got %T", resp.GetBody())
	}
	if init.GetClientHandleId() != 88 || init.GetRootResourceId() != 99 {
		t.Fatalf("expected bus resource service init, got client=%d root=%d", init.GetClientHandleId(), init.GetRootResourceId())
	}
}

type sentinelResourceService struct{}

func (sentinelResourceService) ResourceClient(
	strm resource.SRPCResourceService_ResourceClientStream,
) error {
	// Require an init request before answering with fixed handles.
	req, err := strm.Recv()
	if err != nil {
		return err
	}
	if req.GetInit() == nil {
		return errors.New("expected resource client init")
	}

	// Reply with the sentinel init and hold the stream until it closes.
	if err := strm.Send(&resource.ResourceClientResponse{
		Body: &resource.ResourceClientResponse_Init{
			Init: &resource.ResourceClientInit{
				ClientHandleId: 88,
				RootResourceId: 99,
			},
		},
	}); err != nil {
		return err
	}
	<-strm.Context().Done()
	return context.Canceled
}

func (sentinelResourceService) ResourceRpc(resource.SRPCResourceService_ResourceRpcStream) error {
	return errors.New("unexpected resource rpc")
}

func (sentinelResourceService) ResourceAttach(resource.SRPCResourceService_ResourceAttachStream) error {
	return errors.New("unexpected resource attach")
}