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

✨ Basic http gateway

Browse Source
This commit is contained in:
LittleSheep 2024-10-20 17:42:51 +08:00
parent ff24a43580
commit 9ac6370ecb
6 changed files with 85 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
pkg/http/api/command.go Normal file
View File

@ -0,0 +1,48 @@
package api
import (
"context"
"git.solsynth.dev/hypernet/nexus/pkg/directory"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
"github.com/gofiber/fiber/v2"
"github.com/rs/zerolog/log"
"google.golang.org/grpc/metadata"
"strings"
"time"
)
func invokeCommand(c *fiber.Ctx) error {
command := c.Params("command")
method := strings.ToLower(c.Method())
handler := directory.GetCommandHandler(command, method)
if handler == nil {
return fiber.NewError(fiber.StatusNotFound, "command not found")
}
conn, err := handler.GetGrpcConn()
if err != nil {
return fiber.NewError(fiber.StatusServiceUnavailable, "service unavailable")
}
log.Debug().Str("id", command).Str("method", method).Msg("Invoking command from HTTP Gateway...")
ctx := metadata.AppendToOutgoingContext(c.Context(), "client_id", "http-gateway", "ip", c.IP(), "user_agent", c.Get(fiber.HeaderUserAgent))
ctx, cancel := context.WithTimeout(ctx, time.Second*10)
defer cancel()
out, err := proto.NewCommandControllerClient(conn).SendCommand(ctx, &proto.CommandArgument{
Command: command,
Method: method,
Payload: c.Body(),
})
if err != nil {
return fiber.NewError(fiber.StatusInternalServerError, err.Error())
} else {
if !out.IsDelivered {
log.Debug().Str("id", command).Str("method", method).Msg("Invoking command from HTTP Gateway... failed, delivery not confirmed")
}
return c.Status(int(out.Status)).Send(out.Payload)
}
}

35
pkg/http/api/directory.go
View File

@ -1,13 +1,8 @@
package api
import (
"github.com/spf13/viper"
"strings"
"git.solsynth.dev/hypernet/nexus/pkg/directory"
"github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/proxy"
"github.com/rs/zerolog/log"
"github.com/samber/lo"
)
@ -22,33 +17,3 @@ func listExistsService(c *fiber.Ctx) error {
}
}))
}
func forwardServiceRequest(c *fiber.Ctx) error {
serviceType := c.Params("service")
ogKeyword := serviceType
aliasingMap := viper.GetStringMapString("services.aliases")
if val, ok := aliasingMap[serviceType]; ok {
serviceType = val
}
service := directory.GetServiceInstanceByType(serviceType)
if service == nil || service.HttpAddr == nil {
return fiber.NewError(fiber.StatusNotFound, "service not found")
}
ogUrl := c.Request().URI().String()
url := c.OriginalURL()
url = strings.Replace(url, "/cgi/"+ogKeyword, "/api", 1)
url = "http://" + *service.HttpAddr + url
log.Debug().
Str("from", ogUrl).
Str("to", url).
Str("service", serviceType).
Str("id", service.ID).
Msg("Forwarding request for service...")
return proxy.Do(c, url)
}

2
pkg/http/api/index.go
View File

@ -24,5 +24,5 @@ func MapAPIs(app *fiber.App) {
return c.Next()
}).Get("/ws", websocket.New(listenWebsocket))
app.All("/cgi/:service/*", forwardServiceRequest)
app.All("/cgi/:command", invokeCommand)
}

33
pkg/nex/command.go
View File

@ -9,6 +9,7 @@ import (
"google.golang.org/grpc/reflection"
"net"
"net/http"
"strings"
"time"
)
@ -19,20 +20,33 @@ func GetCommandKey(id, method string) string {
}
func (v *Conn) AddCommand(id, method string, tags []string, fn CommandHandler) error {
method = strings.ToLower(method)
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
var addingMethodQueue []string
if method == "all" {
addingMethodQueue = []string{"get", "post", "put", "patch", "delete"}
} else {
addingMethodQueue = append(addingMethodQueue, method)
}
return err
for _, method := range addingMethodQueue {
ky := GetCommandKey(id, method)
_, err := dir.AddCommand(ctx, &proto.CommandInfo{
Id: id,
Method: method,
Tags: tags,
})
if err == nil {
v.commandHandlers[ky] = fn
} else {
return err
}
}
return nil
}
type localCommandRpcServer struct {
@ -43,7 +57,8 @@ type localCommandRpcServer struct {
}
func (v localCommandRpcServer) SendCommand(ctx context.Context, argument *proto.CommandArgument) (*proto.CommandReturn, error) {
if handler, ok := v.conn.commandHandlers[argument.GetCommand()]; !ok {
ky := GetCommandKey(argument.GetCommand(), argument.GetMethod())
if handler, ok := v.conn.commandHandlers[ky]; !ok {
return &proto.CommandReturn{
Status: http.StatusNotFound,
Payload: []byte(argument.GetCommand() + " not found"),

16
pkg/nex/command_test.go
View File

@ -25,7 +25,15 @@ func TestHandleCommand(t *testing.T) {
t.Fatal(fmt.Errorf("unable to register service: %v", err))
}
err = conn.AddCommand("echo", "get", nil, func(ctx *nex.CommandCtx) error {
err = conn.AddCommand("say.hi", "all", nil, func(ctx *nex.CommandCtx) error {
return ctx.Write([]byte("Hello, World!"), http.StatusOK)
})
if err != nil {
t.Fatal(fmt.Errorf("unable to add command: %v", err))
return
}
err = conn.AddCommand("echo", "all", nil, func(ctx *nex.CommandCtx) error {
t.Log("Received command: ", string(ctx.Read()))
return ctx.Write(ctx.Read(), http.StatusOK)
})
if err != nil {
@ -34,13 +42,13 @@ func TestHandleCommand(t *testing.T) {
}
go func() {
err := conn.RunCommands("127.0.0.1:6001")
err := conn.RunCommands("0.0.0.0: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)
t.Log("Waiting 60 seconds for calling command...")
time.Sleep(time.Second * 60)
}

3
settings.toml
View File

@ -24,8 +24,5 @@ refresh_token_duration = 2592000
dsn = "host=localhost user=postgres password=password dbname=hy_dealer port=5432 sslmode=disable"
prefix = "dealer_"
[services]
aliases = { id = "auth", uc = "files", co = "interactive", im = "messaging" }
[scraper]
user-agent = "SolarBot/1.0"