more tests & refactorings

internal/app/app.go

@@ -49,7 +49,7 @@ func (a *App) Run(ctx context.Context) error {
 	serviceRegistry := messaging.NewServiceRegistry(a.logger)
 	// start rpc client if host is provided, otherwise bot serves as a distributor bot (rpc server)
 	if a.cfg.PartnerPluginConfig.Host != "" {
-		a.startRPCClient(gCtx, g, *serviceRegistry)
+		a.startRPCClient(gCtx, g, serviceRegistry)
 	} else {
 		a.logger.Infof("No host for partner plugin provided, bot will serve as a distributor bot.")
 		serviceRegistry.RegisterServices(a.cfg.SupportedRequestTypes, nil)
@@ -122,7 +122,7 @@ func (a *App) startMessenger(ctx context.Context, g *errgroup.Group) (messaging.
 	return messenger, userIDUpdatedChan
 }
 
-func (a *App) startRPCServer(ctx context.Context, msgProcessor messaging.Processor, serviceRegistry *messaging.ServiceRegistry, g *errgroup.Group) {
+func (a *App) startRPCServer(ctx context.Context, msgProcessor messaging.Processor, serviceRegistry messaging.ServiceRegistry, g *errgroup.Group) {
 	rpcServer := server.NewServer(&a.cfg.RPCServerConfig, a.logger, msgProcessor, serviceRegistry)
 	g.Go(func() error {
 		a.logger.Info("Starting gRPC server...")
@@ -135,7 +135,7 @@ func (a *App) startRPCServer(ctx context.Context, msgProcessor messaging.Process
 	})
 }
 
-func (a *App) startMessageProcessor(ctx context.Context, messenger messaging.Messenger, serviceRegistry *messaging.ServiceRegistry, responseHandler messaging.ResponseHandler, g *errgroup.Group, userIDUpdated chan string) messaging.Processor {
+func (a *App) startMessageProcessor(ctx context.Context, messenger messaging.Messenger, serviceRegistry messaging.ServiceRegistry, responseHandler messaging.ResponseHandler, g *errgroup.Group, userIDUpdated chan string) messaging.Processor {
 	msgProcessor := messaging.NewProcessor(messenger, a.logger, a.cfg.ProcessorConfig, serviceRegistry, responseHandler)
 	g.Go(func() error {
 		// Wait for userID to be passed

internal/messaging/noop_response_handler.go

@@ -13,5 +13,5 @@ var _ ResponseHandler = (*NoopResponseHandler)(nil)
 
 type NoopResponseHandler struct{}
 
-func (n NoopResponseHandler) HandleResponse(context.Context, MessageType, *RequestContent, *ResponseContent) {
+func (NoopResponseHandler) HandleResponse(context.Context, MessageType, *RequestContent, *ResponseContent) {
 }

internal/messaging/processor.go

@@ -14,10 +14,6 @@ import (
 	grpc_metadata "google.golang.org/grpc/metadata"
 )
 
-type InvalidMessageError struct {
-	error
-}
-
 var (
 	_ Processor = (*processor)(nil)
 
@@ -51,7 +47,7 @@ type processor struct {
 
 	mu               sync.Mutex
 	responseChannels map[string]chan *Message
-	serviceRegistry  *ServiceRegistry
+	serviceRegistry  ServiceRegistry
 	responseHandler  ResponseHandler
 }
 
@@ -63,7 +59,7 @@ func (p *processor) Checkpoint() string {
 	return "processor"
 }
 
-func NewProcessor(messenger Messenger, logger *zap.SugaredLogger, cfg config.ProcessorConfig, registry *ServiceRegistry, responseHandler ResponseHandler) Processor {
+func NewProcessor(messenger Messenger, logger *zap.SugaredLogger, cfg config.ProcessorConfig, registry ServiceRegistry, responseHandler ResponseHandler) Processor {
 	return &processor{
 		cfg:              cfg,
 		messenger:        messenger,
@@ -105,7 +101,7 @@ func (p *processor) ProcessInbound(msg *Message) error {
 			p.Forward(msg)
 			return nil
 		default:
-			return InvalidMessageError{ErrUnknownMessageCategory}
+			return ErrUnknownMessageCategory
 		}
 	} else {
 		return nil // ignore own outbound messages
@@ -198,10 +194,10 @@ func (p *processor) Respond(msg *Message) error {
 func (p *processor) Forward(msg *Message) {
 	p.logger.Debugf("Forwarding outbound response message: %s", msg.Metadata.RequestID)
 	p.mu.Lock()
+	defer p.mu.Unlock()
 	responseChan, ok := p.responseChannels[msg.Metadata.RequestID]
 	if ok {
 		responseChan <- msg
 		close(responseChan)
 	}
-	p.mu.Unlock()
 }