Add grpc service register.

Signed-off-by: xuezhaojun <zxue@redhat.com>

Author: xuezhaojun

pkg/cloudevents/server/grpc/options/server.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"net"
 	"os"
 	"sync"
 
@@ -29,15 +30,16 @@ type PreStartHook interface {
 }
 
 type Server struct {
-	options        *GRPCServerOptions
-	authenticators []authn.Authenticator
-	authorizers    []authz.Authorizer
-	services       map[types.CloudEventsDataType]server.Service
-	hooks          []PreStartHook
+	options            *GRPCServerOptions
+	authenticators     []authn.Authenticator
+	authorizers        []authz.Authorizer
+	cloudeventServices map[types.CloudEventsDataType]server.Service
+	grpcService        []func(*grpc.Server)
+	hooks              []PreStartHook
 }
 
 func NewServer(opt *GRPCServerOptions) *Server {
-	return &Server{options: opt, services: make(map[types.CloudEventsDataType]server.Service)}
+	return &Server{options: opt, cloudeventServices: make(map[types.CloudEventsDataType]server.Service)}
 }
 
 func (s *Server) WithAuthenticator(authenticator authn.Authenticator) *Server {
@@ -51,7 +53,7 @@ func (s *Server) WithAuthorizer(authorizer authz.Authorizer) *Server {
 }
 
 func (s *Server) WithService(t types.CloudEventsDataType, service server.Service) *Server {
-	s.services[t] = service
+	s.cloudeventServices[t] = service
 	return s
 }
 
@@ -60,6 +62,11 @@ func (s *Server) WithPreStartHooks(hooks ...PreStartHook) *Server {
 	return s
 }
 
+func (s *Server) WithGRPCService(fn func(*grpc.Server)) *Server {
+	s.grpcService = append(s.grpcService, fn)
+	return s
+}
+
 func (s *Server) Run(ctx context.Context) error {
 	var grpcServerOptions []grpc.ServerOption
 	grpcServerOptions = append(grpcServerOptions, grpc.MaxRecvMsgSize(s.options.MaxReceiveMessageSize))
@@ -116,19 +123,38 @@ func (s *Server) Run(ctx context.Context) error {
 			newAuthzStreamInterceptor(s.authorizers...)))
 
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := grpcserver.NewGRPCBroker(grpcServer)
 
-	for t, service := range s.services {
+	// Register the CloudEventServiceServer
+	grpcEventServer := grpcserver.NewGRPCBroker(grpcServer)
+	for t, service := range s.cloudeventServices {
 		grpcEventServer.RegisterService(t, service)
 	}
 
+	// Register other grpc services
+	for _, fn := range s.grpcService {
+		fn(grpcServer)
+	}
+
 	// start hook
 	for _, hook := range s.hooks {
 		hook.Run(ctx)
 	}
 
-	go grpcEventServer.Start(ctx, ":"+s.options.ServerBindPort)
+	// start grpc server
+	lis, err := net.Listen("tcp", ":"+s.options.ServerBindPort)
+	if err != nil {
+		return fmt.Errorf("failed to listen: %v", err)
+	}
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			klog.Errorf("failed to serve gRPC broker: %v", err)
+		}
+	}()
+
 	<-ctx.Done()
+
+	klog.Infof("Shutting down gRPC server")
+	grpcServer.GracefulStop()
 	return nil
 }
 

pkg/cloudevents/server/grpc/server.go

@@ -4,12 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net"
 	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -78,26 +76,6 @@ func (bkr *GRPCBroker) Subscribers() sets.Set[string] {
 	return subscribers
 }
 
-// Start starts the gRPC broker at the given address
-func (bkr *GRPCBroker) Start(ctx context.Context, addr string) {
-	logger := klog.FromContext(ctx)
-	logger.Info("Starting gRPC broker at addr", "addr", addr)
-	lis, err := net.Listen("tcp", addr)
-	if err != nil {
-		utilruntime.Must(fmt.Errorf("failed to listen: %v", err))
-	}
-	go func() {
-		if err := bkr.grpcServer.Serve(lis); err != nil {
-			utilruntime.Must(fmt.Errorf("failed to serve gRPC broker: %v", err))
-		}
-	}()
-
-	// wait until context is canceled
-	<-ctx.Done()
-	klog.Infof("Shutting down gRPC broker")
-	bkr.grpcServer.GracefulStop()
-}
-
 // Publish in stub implementation for agent publish resource status.
 func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest) (*emptypb.Empty, error) {
 	logger := klog.FromContext(ctx)

pkg/cloudevents/server/interface.go

@@ -17,9 +17,6 @@ type AgentEventServer interface {
 	// RegisterService registers a backend service with a certain data type.
 	RegisterService(t types.CloudEventsDataType, service Service)
 
-	// Start initiates the EventServer to listen to agents.
-	Start(ctx context.Context, addr string)
-
 	// Subscribers returns all current subscribers who subscribe to this server.
 	Subscribers() sets.Set[string]
 }