HyperNet/Nexus
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Command registration

Browse Source
This commit is contained in:
LittleSheep
2024-10-20 17:23:53 +08:00
parent 18ce501089
commit ff24a43580
13 changed files with 335 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
pkg/nex/command.go Normal file
View File

@ -0,0 +1,104 @@
package nex
import (
"context"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
"google.golang.org/grpc"
health "google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/reflection"
"net"
"net/http"
"time"
)
type CommandHandler func(ctx *CommandCtx) error
func GetCommandKey(id, method string) string {
return id + ":" + method
}
func (v *Conn) AddCommand(id, method string, tags []string, fn CommandHandler) error {
dir := proto.NewCommandControllerClient(v.nexusConn)
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "client_id", v.Info.Id)
_, err := dir.AddCommand(ctx, &proto.CommandInfo{
Id: id,
Method: method,
Tags: tags,
})
if err == nil {
v.commandHandlers[GetCommandKey(id, method)] = fn
}
return err
}
type localCommandRpcServer struct {
conn *Conn
proto.UnimplementedCommandControllerServer
health.UnimplementedHealthServer
}
func (v localCommandRpcServer) SendCommand(ctx context.Context, argument *proto.CommandArgument) (*proto.CommandReturn, error) {
if handler, ok := v.conn.commandHandlers[argument.GetCommand()]; !ok {
return &proto.CommandReturn{
Status: http.StatusNotFound,
Payload: []byte(argument.GetCommand() + " not found"),
}, nil
} else {
cc := &CommandCtx{
requestBody: argument.GetPayload(),
statusCode: http.StatusOK,
}
if md, ok := metadata.FromIncomingContext(ctx); ok {
for k, v := range md {
cc.values.Store(k, v)
}
}
if err := handler(cc); err != nil {
return nil, err
} else {
return &proto.CommandReturn{
Status: int32(cc.statusCode),
Payload: cc.responseBody,
}, nil
}
}
}
func (v localCommandRpcServer) Check(ctx context.Context, request *health.HealthCheckRequest) (*health.HealthCheckResponse, error) {
return &health.HealthCheckResponse{
Status: health.HealthCheckResponse_SERVING,
}, nil
}
func (v localCommandRpcServer) Watch(request *health.HealthCheckRequest, server health.Health_WatchServer) error {
for {
if server.Send(&health.HealthCheckResponse{
Status: health.HealthCheckResponse_SERVING,
}) != nil {
break
}
time.Sleep(1000 * time.Millisecond)
}
return nil
}
func (v *Conn) RunCommands(addr string) error {
v.commandServer = grpc.NewServer()
service := &localCommandRpcServer{conn: v}
proto.RegisterCommandControllerServer(v.commandServer, service)
health.RegisterHealthServer(v.commandServer, service)
reflection.Register(v.commandServer)
listener, err := net.Listen("tcp", addr)
if err != nil {
return err
}
return v.commandServer.Serve(listener)
}

47
pkg/nex/command_context.go Normal file
View File

@ -0,0 +1,47 @@
package nex
import (
"github.com/goccy/go-json"
"sync"
)
type CommandCtx struct {
requestBody []byte
responseBody []byte
statusCode int
values sync.Map
}
func (c *CommandCtx) Value(key string, newValue ...any) any {
if len(newValue) > 0 {
c.values.Store(key, newValue[0])
}
val, _ := c.values.Load(key)
return val
}
func (c *CommandCtx) Read() []byte {
return c.requestBody
}
func (c *CommandCtx) ReadJSON(out any) error {
return json.Unmarshal(c.requestBody, out)
}
func (c *CommandCtx) Write(data []byte, statusCode ...int) error {
c.responseBody = data
if len(statusCode) > 0 {
c.statusCode = statusCode[0]
}
return nil
}
func (c *CommandCtx) JSON(data any, statusCode ...int) error {
raw, err := json.Marshal(data)
if err != nil {
return err
}
return c.Write(raw, statusCode...)
}

46
pkg/nex/command_test.go Normal file
View File

@ -0,0 +1,46 @@
package nex_test
import (
"fmt"
"git.solsynth.dev/hypernet/nexus/pkg/nex"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
"net/http"
"testing"
"time"
)
func TestHandleCommand(t *testing.T) {
conn, err := nex.NewNexusConn("127.0.0.1:7001", &proto.ServiceInfo{
Id: "echo01",
Type: "echo",
Label: "Echo",
GrpcAddr: "127.0.0.1:6001",
HttpAddr: nil,
})
if err != nil {
t.Fatal(fmt.Errorf("unable to connect nexus: %v", err))
}
if err := conn.RegisterService(); err != nil {
t.Fatal(fmt.Errorf("unable to register service: %v", err))
}
err = conn.AddCommand("echo", "get", nil, func(ctx *nex.CommandCtx) error {
return ctx.Write(ctx.Read(), http.StatusOK)
})
if err != nil {
t.Fatal(fmt.Errorf("unable to add command: %v", err))
return
}
go func() {
err := conn.RunCommands("127.0.0.1:6001")
if err != nil {
t.Error(fmt.Errorf("unable to run commands: %v", err))
return
}
}()
t.Log("Waiting 10 seconds for calling command...")
time.Sleep(time.Second * 10)
}

48
pkg/nex/conn.go
View File

@ -2,6 +2,7 @@ package nex
import (
"context"
"google.golang.org/grpc/metadata"
"time"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
@ -12,15 +13,18 @@ import (
_ "github.com/mbobakov/grpc-consul-resolver"
)
type HyperConn struct {
type Conn struct {
Addr string
Info *proto.ServiceInfo
dealerConn *grpc.ClientConn
cacheGrpcConn map[string]*grpc.ClientConn
commandServer *grpc.Server
commandHandlers map[string]CommandHandler
nexusConn *grpc.ClientConn
clientConn map[string]*grpc.ClientConn
}
func NewHyperConn(addr string, info *proto.ServiceInfo) (*HyperConn, error) {
func NewNexusConn(addr string, info *proto.ServiceInfo) (*Conn, error) {
conn, err := grpc.NewClient(
addr,
grpc.WithTransportCredentials(insecure.NewCredentials()),
@ -29,22 +33,26 @@ func NewHyperConn(addr string, info *proto.ServiceInfo) (*HyperConn, error) {
return nil, err
}
return &HyperConn{
return &Conn{
Addr: addr,
Info: info,
dealerConn: conn,
cacheGrpcConn: make(map[string]*grpc.ClientConn),
commandHandlers: make(map[string]CommandHandler),
nexusConn: conn,
clientConn: make(map[string]*grpc.ClientConn),
}, nil
}
func (v *HyperConn) RegisterService() error {
dir := proto.NewServiceDirectoryClient(v.dealerConn)
_, err := dir.AddService(context.Background(), v.Info)
func (v *Conn) RegisterService() error {
dir := proto.NewServiceDirectoryClient(v.nexusConn)
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "client_id", v.Info.Id)
_, err := dir.AddService(ctx, v.Info)
return err
}
func (v *HyperConn) KeepRegisterService() error {
func (v *Conn) RunRegistering() error {
err := v.RegisterService()
if err != nil {
return err
@ -52,9 +60,9 @@ func (v *HyperConn) KeepRegisterService() error {
for {
time.Sleep(5 * time.Second)
client := health.NewHealthClient(v.dealerConn)
client := health.NewHealthClient(v.nexusConn)
if _, err := client.Check(context.Background(), &health.HealthCheckRequest{}); err != nil {
if v.KeepRegisterService() == nil {
if v.RunRegistering() == nil {
break
}
}
@ -63,12 +71,12 @@ func (v *HyperConn) KeepRegisterService() error {
return nil
}
func (v *HyperConn) GetNexusGrpcConn() *grpc.ClientConn {
return v.dealerConn
func (v *Conn) GetNexusGrpcConn() *grpc.ClientConn {
return v.nexusConn
}
func (v *HyperConn) GetServiceGrpcConn(t string) (*grpc.ClientConn, error) {
if val, ok := v.cacheGrpcConn[t]; ok {
func (v *Conn) GetClientGrpcConn(t string) (*grpc.ClientConn, error) {
if val, ok := v.clientConn[t]; ok {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
if _, err := health.NewHealthClient(val).Check(ctx, &health.HealthCheckRequest{
@ -76,14 +84,14 @@ func (v *HyperConn) GetServiceGrpcConn(t string) (*grpc.ClientConn, error) {
}); err == nil {
return val, nil
} else {
delete(v.cacheGrpcConn, t)
delete(v.clientConn, t)
}
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
out, err := proto.NewServiceDirectoryClient(v.dealerConn).GetService(ctx, &proto.GetServiceRequest{
out, err := proto.NewServiceDirectoryClient(v.nexusConn).GetService(ctx, &proto.GetServiceRequest{
Type: &t,
})
if err != nil {
@ -95,7 +103,7 @@ func (v *HyperConn) GetServiceGrpcConn(t string) (*grpc.ClientConn, error) {
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
if err == nil {
v.cacheGrpcConn[t] = conn
v.clientConn[t] = conn
}
return conn, err
}