WIP: initial commit

service/option.go

@@ -0,0 +1 @@
+package service

service/serve.go

@@ -0,0 +1,139 @@
+package service
+
+/*
+Copyright 2025 Suyono <suyono3484@gmail.com>
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"sync/atomic"
+	"time"
+
+	pb "gitea.suyono.dev/suyono/go-agent/proto"
+	"gitea.suyono.dev/suyono/go-agent/storage"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+type Option interface {
+	SocketPath() string
+	Expiry() string
+	StopGraceTime() time.Duration
+}
+
+type Storage interface {
+	Get(string) (string, error)
+	Set(string, string) error
+	Shutdown() error
+}
+
+type server struct {
+	pb.UnimplementedAgentServer
+	stor Storage
+}
+
+func (s *server) Get(ctx context.Context, req *pb.CacheGetRequest) (*pb.CacheValue, error) {
+	val, err := s.stor.Get(req.GetKey())
+	if err != nil {
+		return nil, err
+	}
+
+	result := &pb.CacheValue{
+		Status: "OK",
+		Value:  val,
+	}
+	return result, nil
+}
+
+func (s *server) Set(ctx context.Context, req *pb.CacheSetRequest) (*pb.SetStatus, error) {
+	err := s.stor.Set(req.GetKey(), req.GetValue())
+	if err != nil {
+		return nil, err
+	}
+	return &pb.SetStatus{
+		Status: "OK",
+	}, nil
+}
+
+func (s *server) Shutdown(context.Context, *emptypb.Empty) (*emptypb.Empty, error) {
+	if err := s.stor.Shutdown(); err != nil {
+		return nil, err
+	}
+	return &emptypb.Empty{}, nil
+}
+
+func Serve(ctx context.Context, opt Option) error {
+	var (
+		l      net.Listener
+		err    error
+		errC   error
+		ok     bool
+		sw     *atomic.Bool
+		cancel context.CancelFunc
+	)
+	socketPath := opt.SocketPath()
+	os.Remove(socketPath)
+
+	if l, err = net.Listen("unix", socketPath); err != nil {
+		return fmt.Errorf("failed to listen on unix socket: %w", err)
+	}
+
+	grpcServer := grpc.NewServer()
+	pb.RegisterAgentServer(grpcServer, &server{stor: storage.NewInMap()})
+	errChan := make(chan error, 1)
+	ctx, cancel = context.WithCancel(ctx)
+	sw = new(atomic.Bool)
+	sw.Store(false)
+
+	go func() {
+		err := grpcServer.Serve(l)
+		sw.Swap(true)
+		if err != nil {
+			errChan <- err
+		}
+		close(errChan)
+		cancel()
+	}()
+
+	<-ctx.Done()
+	if !sw.Load() {
+		grpcServer.GracefulStop()
+	}
+
+	fst := time.NewTimer(opt.StopGraceTime())
+	err = nil
+fsl:
+	for {
+		select {
+		case errC, ok = <-errChan:
+			if !ok {
+				break fsl
+			}
+			if errC != nil {
+				err = errC
+			}
+		case <-fst.C:
+			if !sw.Load() {
+				grpcServer.Stop()
+			}
+			break fsl
+		}
+	}
+
+	return err
+}