syonad/two
2
0
Fork 0
Code Issues 6 Pull requests 1 Projects Releases 9 Packages Wiki Activity Actions

f-21: refactor: add dispatcher layer for MQTT migration

Browse source 
Introduce internal/dispatcher package with a Command interface and typed
commands (CreateVPC, DeleteVPC, CreateSubnet, DeleteSubnet). The API
handlers now call dispatcher.Dispatch() instead of enqueuing closures
directly, decoupling transport (HTTP today, MQTT tomorrow) from execution.

Signed-off-by: GnomeZworc <nicolas.boufidjeline@g3e.fr>
This commit is contained in:
GnomeZworc 2026-04-19 00:06:50 +02:00
commit 63a288f69e
Signed by: nicolas.boufideline
GPG key ID: 4406BBBF8845D632
9 changed files with 114 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

4
cmd/agent/main.go
View file

@ -7,6 +7,7 @@ import (
agentapi "git.g3e.fr/syonad/two/internal/api/agent"
configuration "git.g3e.fr/syonad/two/internal/config/agent"
"git.g3e.fr/syonad/two/internal/dispatcher"
agentmetrics "git.g3e.fr/syonad/two/internal/prometheus/agent"
"git.g3e.fr/syonad/two/pkg/db/kv"
promserver "git.g3e.fr/syonad/two/pkg/prometheus"
@ -35,7 +36,8 @@ func main() {
apiAddr := fmt.Sprintf("%s:%d", cfg.Api.Address, cfg.Api.Port)
promAddr := fmt.Sprintf("%s:%d", cfg.Prometheus.Address, cfg.Prometheus.Port)
go agentapi.New(q, db).Start(apiAddr)
d := dispatcher.New(q, db)
go agentapi.New(d, db).Start(apiAddr)
go promserver.Start(promAddr, registry)
select {}

8
internal/api/agent/server.go
View file

@ -4,17 +4,17 @@ import (
"log"
"net/http"
"git.g3e.fr/syonad/two/pkg/worker"
"git.g3e.fr/syonad/two/internal/dispatcher"
"github.com/dgraph-io/badger/v4"
)
type Server struct {
queue *worker.Queue
dispatcher *dispatcher.Dispatcher
db *badger.DB
}
func New(queue *worker.Queue, db *badger.DB) *Server {
return &Server{queue: queue, db: db}
func New(d *dispatcher.Dispatcher, db *badger.DB) *Server {
return &Server{dispatcher: d, db: db}
}
func (s *Server) Start(address string) {

7
internal/api/agent/subnet.go
View file

@ -4,6 +4,8 @@ import (
"encoding/json"
"net/http"
"strings"
"git.g3e.fr/syonad/two/internal/dispatcher"
)
func (s *Server) SubnetByNameHandler(w http.ResponseWriter, r *http.Request) {
@ -31,11 +33,8 @@ func (s *Server) getSubnet(w http.ResponseWriter, r *http.Request, name string)
}
func (s *Server) deleteSubnet(w http.ResponseWriter, r *http.Request, name string) {
s.queue.Submit(func() {
destroySubnet(name)
})
s.dispatcher.Dispatch(dispatcher.DeleteSubnetCommand{Name: name})
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(Subnet{Name: name, State: "deleting"})
}
func destroySubnet(name string) {}

12
internal/api/agent/subnets.go
View file

@ -3,6 +3,8 @@ package agentapi
import (
"encoding/json"
"net/http"
"git.g3e.fr/syonad/two/internal/dispatcher"
)
func (s *Server) SubnetsHandler(w http.ResponseWriter, r *http.Request) {
@ -34,8 +36,13 @@ func (s *Server) postSubnet(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(ErrorResponse{Error: "name, vpc, local_ip, gateway_ip and cidr are required"})
return
}
s.queue.Submit(func() {
createSubnet(req)
s.dispatcher.Dispatch(dispatcher.CreateSubnetCommand{
Name: req.Name,
VPC: req.VPC,
VxlanID: req.VxlanID,
LocalIP: req.LocalIP,
GatewayIP: req.GatewayIP,
CIDR: req.CIDR,
})
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(Subnet{
@ -49,4 +56,3 @@ func (s *Server) postSubnet(w http.ResponseWriter, r *http.Request) {
})
}
func createSubnet(req SubnetCreateRequest) {}

19
internal/api/agent/vpc.go
View file

@ -2,13 +2,10 @@ package agentapi
import (
"encoding/json"
"fmt"
"net/http"
"os"
"strings"
"git.g3e.fr/syonad/two/internal/vpc"
"git.g3e.fr/syonad/two/pkg/db/kv"
"git.g3e.fr/syonad/two/internal/dispatcher"
)
func (s *Server) VpcByNameHandler(w http.ResponseWriter, r *http.Request) {
@ -36,20 +33,8 @@ func (s *Server) getVpc(w http.ResponseWriter, _ *http.Request, name string) {
}
func (s *Server) deleteVpc(w http.ResponseWriter, _ *http.Request, name string) {
s.queue.Submit(func() {
kv.AddInDB(s.db, "vpc/"+name+"/state", "deleting")
if err := vpc.DeleteVPC(s.db, name); err != nil {
fmt.Println(err)
}
if state, err := kv.GetFromDB(s.db, "vpc/"+name+"/state"); err != nil {
fmt.Println(err)
os.Exit(1)
} else if state == "deleted" {
kv.DeleteInDB(s.db, "vpc/"+name)
}
})
s.dispatcher.Dispatch(dispatcher.DeleteVPCCommand{Name: name})
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(VPC{Name: name, State: "deleting"})
}
func destroyVpc(name string) {}

11
internal/api/agent/vpcs.go
View file

@ -2,11 +2,9 @@ package agentapi
import (
"encoding/json"
"fmt"
"net/http"
"git.g3e.fr/syonad/two/internal/vpc"
"git.g3e.fr/syonad/two/pkg/db/kv"
"git.g3e.fr/syonad/two/internal/dispatcher"
)
func (s *Server) VpcsHandler(w http.ResponseWriter, r *http.Request) {
@ -38,12 +36,7 @@ func (s *Server) postVpc(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(ErrorResponse{Error: "name is required"})
return
}
s.queue.Submit(func() {
kv.AddInDB(s.db, "vpc/"+req.Name+"/state", "creating")
if err := vpc.CreateVPC(s.db, req.Name); err != nil {
fmt.Println(err)
}
})
s.dispatcher.Dispatch(dispatcher.CreateVPCCommand{Name: req.Name})
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(VPC{Name: req.Name, State: "creating"})
}

29
internal/dispatcher/dispatcher.go Normal file
View file

@ -0,0 +1,29 @@
package dispatcher
import (
"log"
"git.g3e.fr/syonad/two/pkg/worker"
"github.com/dgraph-io/badger/v4"
)
type Command interface {
Execute(db *badger.DB) error
}
type Dispatcher struct {
queue *worker.Queue
db *badger.DB
}
func New(queue *worker.Queue, db *badger.DB) *Dispatcher {
return &Dispatcher{queue: queue, db: db}
}
func (d *Dispatcher) Dispatch(cmd Command) {
d.queue.Submit(func() {
if err := cmd.Execute(d.db); err != nil {
log.Printf("command error (%T): %v", cmd, err)
}
})
}

26
internal/dispatcher/subnet_commands.go Normal file
View file

@ -0,0 +1,26 @@
package dispatcher
import "github.com/dgraph-io/badger/v4"
type CreateSubnetCommand struct {
Name string
VPC string
VxlanID int
LocalIP string
GatewayIP string
CIDR string
}
func (c CreateSubnetCommand) Execute(db *badger.DB) error {
// TODO: brancher internal/subnet/create.go
return nil
}
type DeleteSubnetCommand struct {
Name string
}
func (c DeleteSubnetCommand) Execute(db *badger.DB) error {
// TODO: brancher internal/subnet/delete.go
return nil
}

35
internal/dispatcher/vpc_commands.go Normal file
View file

@ -0,0 +1,35 @@
package dispatcher
import (
"git.g3e.fr/syonad/two/internal/vpc"
"git.g3e.fr/syonad/two/pkg/db/kv"
"github.com/dgraph-io/badger/v4"
)
type CreateVPCCommand struct {
Name string
}
func (c CreateVPCCommand) Execute(db *badger.DB) error {
kv.AddInDB(db, "vpc/"+c.Name+"/state", "creating")
return vpc.CreateVPC(db, c.Name)
}
type DeleteVPCCommand struct {
Name string
}
func (c DeleteVPCCommand) Execute(db *badger.DB) error {
kv.AddInDB(db, "vpc/"+c.Name+"/state", "deleting")
if err := vpc.DeleteVPC(db, c.Name); err != nil {
return err
}
state, err := kv.GetFromDB(db, "vpc/"+c.Name+"/state")
if err != nil {
return err
}
if state == "deleted" {
kv.DeleteInDB(db, "vpc/"+c.Name)
}
return nil
}