added sync runner and refactored operator creation #171
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,4 @@ | ||
| package rpc | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| package rpc | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "net" | ||
| "sync" | ||
| "sync/atomic" | ||
|
|
||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/codes" | ||
| "google.golang.org/grpc/status" | ||
|
|
||
| slogctx "github.com/veqryn/slog-context" | ||
|
|
||
| "github.com/openkcm/orbital" | ||
| "github.com/openkcm/orbital/codec" | ||
| orbitalv1 "github.com/openkcm/orbital/proto/orbital/v1" | ||
| ) | ||
|
|
||
| var _ orbital.SyncResponder = (*Server)(nil) | ||
|
|
||
| var ( | ||
| // ErrNilListener is returned by NewServer when given a nil net.Listener. | ||
| ErrNilListener = errors.New("grpc server: listener cannot be nil") | ||
| // ErrServerAlreadyRan is returned by Run if the server has already been started. | ||
| ErrServerAlreadyRan = errors.New("grpc server: Run already called") | ||
| ) | ||
|
|
||
| type ( | ||
| // ServerOption configures a Server. | ||
| ServerOption func(*serverConfig) error | ||
|
|
||
| serverConfig struct { | ||
| grpcServerOpts []grpc.ServerOption | ||
| } | ||
|
|
||
| // Server implements the orbital TaskService gRPC server and the | ||
| // SyncResponder interface. It receives task requests over gRPC, | ||
| // processes them via the TaskRequestHandler provided to Run, | ||
| // and returns responses synchronously. | ||
| Server struct { | ||
| orbitalv1.UnimplementedTaskServiceServer | ||
|
|
||
| config serverConfig | ||
| lis net.Listener | ||
| grpcServer *grpc.Server | ||
| handler orbital.TaskRequestHandler | ||
| ran atomic.Bool | ||
| stopOnce sync.Once | ||
| } | ||
| ) | ||
|
|
||
| // NewServer creates a gRPC Server bound to the given listener. | ||
| func NewServer(lis net.Listener, opts ...ServerOption) (*Server, error) { | ||
| if lis == nil { | ||
| return nil, ErrNilListener | ||
| } | ||
|
|
||
| cfg := serverConfig{} | ||
| for _, opt := range opts { | ||
| if err := opt(&cfg); err != nil { | ||
| return nil, err | ||
| } | ||
| } | ||
|
|
||
| return &Server{ | ||
| config: cfg, | ||
| lis: lis, | ||
| }, nil | ||
| } | ||
|
|
||
| // WithServerOptions appends gRPC server options (interceptors, TLS | ||
| // credentials, etc.) used when the underlying grpc.Server is created. | ||
| func WithServerOptions(opts ...grpc.ServerOption) ServerOption { | ||
|
There was a problem hiding this comment.
I think the pkg naming isn't optimal because we now have 2 grpc pkgs using the same name. So using this function requires one to make aliased imports. What if we call this pkg |
||
| return func(cfg *serverConfig) error { | ||
| cfg.grpcServerOpts = append(cfg.grpcServerOpts, opts...) | ||
| return nil | ||
| } | ||
| } | ||
|
|
||
| // Run registers the TaskService, starts serving, and blocks until | ||
|
Contributor
There was a problem hiding this comment.
Contributor
Author
There was a problem hiding this comment. Fixed it so that Run and stop can only be called once on a grpc client. |
||
| // Close is called or ctx is cancelled. Run may only be called once. | ||
| func (s *Server) Run(ctx context.Context, handler orbital.TaskRequestHandler) error { | ||
| if !s.ran.CompareAndSwap(false, true) { | ||
| return ErrServerAlreadyRan | ||
| } | ||
|
|
||
| s.handler = handler | ||
| s.grpcServer = grpc.NewServer(s.config.grpcServerOpts...) | ||
| orbitalv1.RegisterTaskServiceServer(s.grpcServer, s) | ||
|
|
||
| go func() { | ||
| <-ctx.Done() | ||
| s.stop() | ||
| }() | ||
|
|
||
| slogctx.Info(ctx, "grpc server starting", "address", s.lis.Addr().String()) | ||
| return s.grpcServer.Serve(s.lis) | ||
| } | ||
|
|
||
| // Close gracefully stops the gRPC server. If ctx expires before graceful | ||
| // shutdown completes, it force-stops. | ||
| func (s *Server) Close(ctx context.Context) error { | ||
| if s.grpcServer == nil { | ||
| return nil | ||
| } | ||
|
|
||
| stopped := make(chan struct{}) | ||
| go func() { | ||
| s.stop() | ||
| close(stopped) | ||
| }() | ||
|
|
||
| select { | ||
| case <-stopped: | ||
| case <-ctx.Done(): | ||
| s.grpcServer.Stop() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // SendTaskRequest implements orbitalv1.TaskServiceServer. | ||
| func (s *Server) SendTaskRequest(ctx context.Context, pReq *orbitalv1.TaskRequest) (*orbitalv1.TaskResponse, error) { | ||
| req, err := codec.FromProtoToTaskRequest(pReq) | ||
| if err != nil { | ||
| return nil, status.Errorf(codes.InvalidArgument, "invalid task request: %v", err) | ||
| } | ||
|
|
||
| resp, err := s.handler(ctx, req) | ||
| if err != nil { | ||
| return nil, mapProcessError(err) | ||
| } | ||
|
|
||
| return codec.FromTaskResponseToProto(resp), nil | ||
| } | ||
|
|
||
| func mapProcessError(err error) error { | ||
| switch { | ||
| case errors.Is(err, orbital.ErrSignatureInvalid): | ||
| return status.Error(codes.Unauthenticated, err.Error()) | ||
| case errors.Is(err, orbital.ErrResponseSigning): | ||
| return status.Error(codes.Internal, err.Error()) | ||
| default: | ||
| return status.Error(codes.Internal, err.Error()) | ||
| } | ||
| } | ||
|
|
||
| func (s *Server) stop() { | ||
| s.stopOnce.Do(func() { | ||
| if s.grpcServer != nil { | ||
| s.grpcServer.GracefulStop() | ||
| } | ||
| }) | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since this directory evolved quite a bit, we could think of renaming it from
clienttotransportor something else (probable better in a separate PR if at all)