Merge pull request #104 from nitrictech/fix/plugin-checks

Re-instate plugin registration checks for adapters.
nitrictech · Aug 2, 2021 · 04eb070 · 04eb070
2 parents 003f0de + e749464
commit 04eb070
Show file tree

Hide file tree

Showing 6 changed files with 107 additions and 1 deletion.
diff --git a/pkg/adapters/grpc/document_grpc.go b/pkg/adapters/grpc/document_grpc.go
@@ -30,7 +30,19 @@ type DocumentServiceServer struct {
 	documentPlugin document.DocumentService
 }
 
+func (s *DocumentServiceServer) checkPluginRegistered() error {
+	if s.documentPlugin == nil {
+		return NewPluginNotRegisteredError("Document")
+	}
+
+	return nil
+}
+
 func (s *DocumentServiceServer) Get(ctx context.Context, req *pb.DocumentGetRequest) (*pb.DocumentGetResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	key := keyFromWire(req.Key)
 
 	doc, err := s.documentPlugin.Get(key)
@@ -49,6 +61,10 @@ func (s *DocumentServiceServer) Get(ctx context.Context, req *pb.DocumentGetRequ
 }
 
 func (s *DocumentServiceServer) Set(ctx context.Context, req *pb.DocumentSetRequest) (*pb.DocumentSetResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	key := keyFromWire(req.Key)
 
 	err := s.documentPlugin.Set(key, req.GetContent().AsMap())
@@ -60,6 +76,10 @@ func (s *DocumentServiceServer) Set(ctx context.Context, req *pb.DocumentSetRequ
 }
 
 func (s *DocumentServiceServer) Delete(ctx context.Context, req *pb.DocumentDeleteRequest) (*pb.DocumentDeleteResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	key := keyFromWire(req.Key)
 
 	err := s.documentPlugin.Delete(key)
@@ -71,6 +91,10 @@ func (s *DocumentServiceServer) Delete(ctx context.Context, req *pb.DocumentDele
 }
 
 func (s *DocumentServiceServer) Query(ctx context.Context, req *pb.DocumentQueryRequest) (*pb.DocumentQueryResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	collection := collectionFromWire(req.Collection)
 	expressions := make([]document.QueryExpression, len(req.GetExpressions()))
 	for i, exp := range req.GetExpressions() {

diff --git a/pkg/adapters/grpc/utils.go → pkg/adapters/grpc/errors.go b/pkg/adapters/grpc/utils.go → pkg/adapters/grpc/errors.go
@@ -28,3 +28,8 @@ func NewGrpcError(operation string, err error) error {
 		return status.Errorf(codes.Internal, "%s: %v", operation, err)
 	}
 }
+
+// Provides generic error for unregistered plugins
+func NewPluginNotRegisteredError(plugin string) error {
+	return status.Errorf(codes.Unimplemented, "%s plugin not registered", plugin)
+}
diff --git a/pkg/adapters/grpc/utils_test.go → pkg/adapters/grpc/errors_test.go b/pkg/adapters/grpc/utils_test.go → pkg/adapters/grpc/errors_test.go
@@ -23,7 +23,7 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-var _ = Describe("GRPC Utils", func() {
+var _ = Describe("GRPC Errors", func() {
 	Context("GrpcError", func() {
 		When("sdk.IllegalArgumentError", func() {
 			It("Should report GRPC IllegalArgument error", func() {
@@ -40,4 +40,13 @@ var _ = Describe("GRPC Utils", func() {
 			})
 		})
 	})
+
+	Context("PluginNotRegisteredError", func() {
+		When("Creating a New PluginNotRegisteredError", func() {
+			It("Should contain the name of the plugin", func() {
+				err := grpc.NewPluginNotRegisteredError("Document")
+				Expect(err.Error()).To(ContainSubstring("rpc error: code = Unimplemented desc = Document plugin not registered"))
+			})
+		})
+	})
 })
diff --git a/pkg/adapters/grpc/event_grpc.go b/pkg/adapters/grpc/event_grpc.go
@@ -28,7 +28,19 @@ type EventServiceServer struct {
 	eventPlugin events.EventService
 }
 
+func (s *EventServiceServer) checkPluginRegistered() error {
+	if s.eventPlugin == nil {
+		return NewPluginNotRegisteredError("Event")
+	}
+
+	return nil
+}
+
 func (s *EventServiceServer) Publish(ctx context.Context, req *pb.EventPublishRequest) (*pb.EventPublishResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	// auto generate an ID if we did not receive one
 	var ID = req.GetEvent().GetId()
 	if ID == "" {
@@ -60,7 +72,19 @@ type TopicServiceServer struct {
 	eventPlugin events.EventService
 }
 
+func (s *TopicServiceServer) checkPluginRegistered() error {
+	if s.eventPlugin == nil {
+		return NewPluginNotRegisteredError("Event")
+	}
+
+	return nil
+}
+
 func (s *TopicServiceServer) List(context.Context, *pb.TopicListRequest) (*pb.TopicListResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	if res, err := s.eventPlugin.ListTopics(); err == nil {
 		topics := make([]*pb.NitricTopic, len(res))
 		for i, topicName := range res {

diff --git a/pkg/adapters/grpc/queue_grpc.go b/pkg/adapters/grpc/queue_grpc.go
@@ -28,7 +28,19 @@ type QueueServiceServer struct {
 	plugin queue.QueueService
 }
 
+func (s *QueueServiceServer) checkPluginRegistered() error {
+	if s.plugin == nil {
+		return NewPluginNotRegisteredError("Queue")
+	}
+
+	return nil
+}
+
 func (s *QueueServiceServer) Send(ctx context.Context, req *pb.QueueSendRequest) (*pb.QueueSendResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	task := req.GetTask()
 
 	nitricTask := queue.NitricTask{
@@ -46,6 +58,10 @@ func (s *QueueServiceServer) Send(ctx context.Context, req *pb.QueueSendRequest)
 }
 
 func (s *QueueServiceServer) SendBatch(ctx context.Context, req *pb.QueueSendBatchRequest) (*pb.QueueSendBatchResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	// Translate tasks
 	tasks := make([]queue.NitricTask, len(req.GetTasks()))
 	for i, task := range req.GetTasks() {
@@ -78,6 +94,10 @@ func (s *QueueServiceServer) SendBatch(ctx context.Context, req *pb.QueueSendBat
 }
 
 func (s *QueueServiceServer) Receive(ctx context.Context, req *pb.QueueReceiveRequest) (*pb.QueueReceiveResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	// Convert gRPC request to plugin params
 	depth := uint32(req.GetDepth())
 	popOptions := queue.ReceiveOptions{
@@ -111,6 +131,10 @@ func (s *QueueServiceServer) Receive(ctx context.Context, req *pb.QueueReceiveRe
 }
 
 func (s *QueueServiceServer) Complete(ctx context.Context, req *pb.QueueCompleteRequest) (*pb.QueueCompleteResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	// Convert gRPC request to plugin params
 	queueName := req.GetQueue()
 	leaseId := req.GetLeaseId()

diff --git a/pkg/adapters/grpc/storage_grpc.go b/pkg/adapters/grpc/storage_grpc.go
@@ -27,7 +27,19 @@ type StorageServiceServer struct {
 	storagePlugin storage.StorageService
 }
 
+func (s *StorageServiceServer) checkPluginRegistered() error {
+	if s.storagePlugin == nil {
+		return NewPluginNotRegisteredError("Storage")
+	}
+
+	return nil
+}
+
 func (s *StorageServiceServer) Write(ctx context.Context, req *pb.StorageWriteRequest) (*pb.StorageWriteResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	if err := s.storagePlugin.Write(req.GetBucketName(), req.GetKey(), req.GetBody()); err == nil {
 		return &pb.StorageWriteResponse{}, nil
 	} else {
@@ -36,6 +48,10 @@ func (s *StorageServiceServer) Write(ctx context.Context, req *pb.StorageWriteRe
 }
 
 func (s *StorageServiceServer) Read(ctx context.Context, req *pb.StorageReadRequest) (*pb.StorageReadResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	if object, err := s.storagePlugin.Read(req.GetBucketName(), req.GetKey()); err == nil {
 		return &pb.StorageReadResponse{
 			Body: object,
@@ -46,6 +62,10 @@ func (s *StorageServiceServer) Read(ctx context.Context, req *pb.StorageReadRequ
 }
 
 func (s *StorageServiceServer) Delete(ctx context.Context, req *pb.StorageDeleteRequest) (*pb.StorageDeleteResponse, error) {
+	if err := s.checkPluginRegistered(); err != nil {
+		return nil, err
+	}
+
 	if err := s.storagePlugin.Delete(req.GetBucketName(), req.GetKey()); err == nil {
 		return &pb.StorageDeleteResponse{}, nil
 	} else {