hyung-hwan/hodu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

separating http handler to separate structs

Browse Source 
using the pattern supported by http.ServeMux since go 1.22
This commit is contained in:
hyung-hwan 2024-11-25 19:46:18 +09:00
parent 903e4cf6d3
commit dcdadbeb20
8 changed files with 261 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Makefile
View File

@ -1,13 +1,14 @@
SRCS=\ SRCS=\
c-peer.go \
client.go \ client.go \
client-ctl.go \
client-peer.go \
frame.go \ frame.go \
hodu.go \ hodu.go \
hodu.pb.go \ hodu.pb.go \
hodu_grpc.pb.go \ hodu_grpc.pb.go \
packet.go \ packet.go \
s-peer.go \
server.go \ server.go \
server-peer.go \
cmd/main.go cmd/main.go
all: hodu all: hodu

88
client-ctl.go Normal file
View File

@ -0,0 +1,88 @@
package hodu
import "encoding/json"
import "fmt"
import "net/http"
/*
* POST GET PUT DELETE
* /servers - create new server list all servers bulk update delete all servers
* /servers/1 - X get server 1 details update server 1 delete server 1
* /servers/1/xxx -
*/
type client_ctl_servers struct {
c *Client
}
type client_ctl_servers_id struct {
c *Client
}
type client_ctl_clients struct {
c *Client
}
type client_ctl_clients_id struct {
c *Client
}
func (ctl *client_ctl_servers) ServeHTTP(w http.ResponseWriter, req *http.Request) {
var c *Client
var err error
var ptn string
c = ctl.c
_, ptn = c.mux.Handler(req);
fmt.Printf("%s %s %s [%s]\n", ptn, req.Method, req.URL.String(), req.PathValue("id"))
switch req.Method {
case http.MethodGet:
goto bad_request // TODO:
case http.MethodPost:
var s ClientCtlParamServer
var cc ClientConfig
err = json.NewDecoder(req.Body).Decode(&s)
if err != nil {
fmt.Printf ("failed to decode body - %s\n", err.Error())
goto bad_request
}
cc.ServerAddr = s.ServerAddr
cc.PeerAddrs = s.PeerAddrs
c.StartService(&cc) // TODO: this can be blocking. do we have to resolve addresses before calling this? also not good because resolution succeed or fail at each attempt. however ok as ServeHTTP itself is in a goroutine?
w.WriteHeader(http.StatusCreated)
case http.MethodPut:
goto bad_request // TODO:
case http.MethodDelete:
var cts *ClientConn
c.cts_mtx.Lock()
for _, cts = range c.cts_map { cts.ReqStop() }
c.cts_mtx.Unlock()
}
return
bad_request:
w.WriteHeader(http.StatusBadRequest)
return
}
func (ctl *client_ctl_servers_id) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}
func (ctl *client_ctl_clients) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}
func (ctl *client_ctl_clients_id) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}

4
c-peer.go → client-peer.go
View File

@ -4,7 +4,7 @@ import "fmt"
import "net" import "net"
import "sync" import "sync"
func NewClientPeerConn(r *ClientRoute, c *net.TCPConn, id uint32) (*ClientPeerConn) { func NewClientPeerConn(r *ClientRoute, c *net.TCPConn, id uint32) *ClientPeerConn {
var cpc ClientPeerConn var cpc ClientPeerConn
cpc.route = r cpc.route = r
@ -56,7 +56,7 @@ func (cpc *ClientPeerConn) ReqStop() {
} }
} }
func (cpc* ClientPeerConn) CloseWrite() { func (cpc *ClientPeerConn) CloseWrite() {
if cpc.server_peer_eof.CompareAndSwap(false, true) { if cpc.server_peer_eof.CompareAndSwap(false, true) {
if cpc.conn != nil { if cpc.conn != nil {
cpc.conn.CloseWrite() cpc.conn.CloseWrite()

219
client.go
View File

@ -1,29 +1,25 @@
package hodu package hodu
//import "bufio"
import "context" import "context"
import "crypto/tls" import "crypto/tls"
import "encoding/json"
import "errors" import "errors"
import "fmt" import "fmt"
import "math/rand"
import "net" import "net"
import "net/http" import "net/http"
import "sync" import "sync"
import "sync/atomic" import "sync/atomic"
import "time" import "time"
//import "github.com/google/uuid"
import "google.golang.org/grpc" import "google.golang.org/grpc"
import "google.golang.org/grpc/codes" import "google.golang.org/grpc/codes"
import "google.golang.org/grpc/credentials/insecure" import "google.golang.org/grpc/credentials/insecure"
import "google.golang.org/grpc/status" import "google.golang.org/grpc/status"
const PTC_LIMIT = 8192
type PacketStreamClient grpc.BidiStreamingClient[Packet, Packet] type PacketStreamClient grpc.BidiStreamingClient[Packet, Packet]
type ServerConnMap = map[net.Addr]*ServerConn type ClientConnMap = map[net.Addr]*ClientConn
type ClientConnMapById = map[uint32]*ClientConn
type ClientPeerConnMap = map[uint32]*ClientPeerConn type ClientPeerConnMap = map[uint32]*ClientPeerConn
type ClientRouteMap = map[uint32]*ClientRoute type ClientRouteMap = map[uint32]*ClientRoute
type ClientPeerCancelFuncMap = map[uint32]context.CancelFunc type ClientPeerCancelFuncMap = map[uint32]context.CancelFunc
@ -43,13 +39,15 @@ type Client struct {
ctl *http.Server // control server ctl *http.Server // control server
cts_mtx sync.Mutex cts_mtx sync.Mutex
cts_map ServerConnMap cts_map ClientConnMap
cts_map_by_id ClientConnMapById
wg sync.WaitGroup wg sync.WaitGroup
stop_req atomic.Bool stop_req atomic.Bool
stop_chan chan bool stop_chan chan bool
log Logger log Logger
mux *http.ServeMux
} }
type ClientPeerConn struct { type ClientPeerConn struct {
@ -66,10 +64,12 @@ type ClientPeerConn struct {
} }
// client connection to server // client connection to server
type ServerConn struct { type ClientConn struct {
cli *Client cli *Client
cfg *ClientConfig cfg *ClientConfig
saddr *net.TCPAddr // server address that is connected to saddr *net.TCPAddr // server address that is connected to
id uint32
lid string
conn *grpc.ClientConn // grpc connection to the server conn *grpc.ClientConn // grpc connection to the server
hdc HoduClient hdc HoduClient
@ -87,7 +87,7 @@ type ServerConn struct {
} }
type ClientRoute struct { type ClientRoute struct {
cts *ServerConn cts *ClientConn
id uint32 id uint32
peer_addr *net.TCPAddr peer_addr *net.TCPAddr
proto ROUTE_PROTO proto ROUTE_PROTO
@ -130,7 +130,7 @@ func (g *GuardedPacketStreamClient) Context() context.Context {
}*/ }*/
// -------------------------------------------------------------------- // --------------------------------------------------------------------
func NewClientRoute(cts *ServerConn, id uint32, addr *net.TCPAddr, proto ROUTE_PROTO) *ClientRoute { func NewClientRoute(cts *ClientConn, id uint32, addr *net.TCPAddr, proto ROUTE_PROTO) *ClientRoute {
var r ClientRoute var r ClientRoute
r.cts = cts r.cts = cts
@ -152,16 +152,17 @@ func (r *ClientRoute) RunTask(wg *sync.WaitGroup) {
// most useful works are triggered by ReportEvent() and done by ConnectToPeer() // most useful works are triggered by ReportEvent() and done by ConnectToPeer()
defer wg.Done() defer wg.Done()
r.cts.cli.log.Write("", LOG_DEBUG, "Sending route-start for id=%d peer=%s to %s", r.id, r.peer_addr.String(), r.cts.saddr.String())
err = r.cts.psc.Send(MakeRouteStartPacket(r.id, r.proto, r.peer_addr.String())) err = r.cts.psc.Send(MakeRouteStartPacket(r.id, r.proto, r.peer_addr.String()))
if err != nil { if err != nil {
//return fmt.Errorf("unable to send route-start packet - %s", err.Error()) r.cts.cli.log.Write("", LOG_DEBUG, "Failed to Send route-start for id=%d peer=%s to %s", r.id, r.peer_addr.String(), r.cts.saddr.String())
goto done; goto done
} }
main_loop: main_loop:
for { for {
select { select {
case <- r.stop_chan: case <-r.stop_chan:
break main_loop break main_loop
} }
} }
@ -170,9 +171,10 @@ done:
r.ReqStop() r.ReqStop()
r.ptc_wg.Wait() // wait for all peer tasks are finished r.ptc_wg.Wait() // wait for all peer tasks are finished
r.cts.cli.log.Write("", LOG_DEBUG, "Sending route-stop for id=%d peer=%s to %s", r.id, r.peer_addr.String(), r.cts.saddr.String())
r.cts.psc.Send(MakeRouteStopPacket(r.id, r.proto, r.peer_addr.String())) r.cts.psc.Send(MakeRouteStopPacket(r.id, r.proto, r.peer_addr.String()))
r.cts.RemoveClientRoute(r) r.cts.RemoveClientRoute(r)
fmt.Printf ("*** End fo Client Roue Task\n") fmt.Printf("*** End fo Client Roue Task\n")
} }
func (r *ClientRoute) ReqStop() { func (r *ClientRoute) ReqStop() {
@ -183,10 +185,10 @@ func (r *ClientRoute) ReqStop() {
} }
r.stop_chan <- true r.stop_chan <- true
} }
fmt.Printf ("*** Sent stop request to Route..\n") fmt.Printf("*** Sent stop request to Route..\n")
} }
func (r* ClientRoute) ConnectToPeer(pts_id uint32, wg *sync.WaitGroup) { func (r *ClientRoute) ConnectToPeer(pts_id uint32, wg *sync.WaitGroup) {
var err error var err error
var conn net.Conn var conn net.Conn
var real_conn *net.TCPConn var real_conn *net.TCPConn
@ -214,7 +216,7 @@ func (r* ClientRoute) ConnectToPeer(pts_id uint32, wg *sync.WaitGroup) {
if err != nil { if err != nil {
// TODO: make send peer started failure mesage? // TODO: make send peer started failure mesage?
fmt.Printf ("failed to connect to %s - %s\n", r.peer_addr.String(), err.Error()) fmt.Printf("failed to connect to %s - %s\n", r.peer_addr.String(), err.Error())
goto peer_aborted goto peer_aborted
} }
@ -252,7 +254,7 @@ peer_aborted:
} }
} }
func (r* ClientRoute) DisconnectFromPeer(pts_id uint32) error { func (r *ClientRoute) DisconnectFromPeer(pts_id uint32) error {
var ptc *ClientPeerConn var ptc *ClientPeerConn
var cancel context.CancelFunc var cancel context.CancelFunc
var ok bool var ok bool
@ -260,7 +262,7 @@ func (r* ClientRoute) DisconnectFromPeer(pts_id uint32) error {
r.ptc_mtx.Lock() r.ptc_mtx.Lock()
cancel, ok = r.ptc_cancel_map[pts_id] cancel, ok = r.ptc_cancel_map[pts_id]
if ok { if ok {
fmt.Printf ("~~~~~~~~~~~~~~~~ cancelling.....\n") fmt.Printf("~~~~~~~~~~~~~~~~ cancelling.....\n")
cancel() cancel()
} }
@ -275,7 +277,7 @@ fmt.Printf ("~~~~~~~~~~~~~~~~ cancelling.....\n")
return nil return nil
} }
func (r* ClientRoute) CloseWriteToPeer(pts_id uint32) error { func (r *ClientRoute) CloseWriteToPeer(pts_id uint32) error {
var ptc *ClientPeerConn var ptc *ClientPeerConn
var ok bool var ok bool
@ -291,27 +293,26 @@ func (r* ClientRoute) CloseWriteToPeer(pts_id uint32) error {
return nil return nil
} }
func (r *ClientRoute) ReportEvent(pts_id uint32, event_type PACKET_KIND, event_data []byte) error {
func (r* ClientRoute) ReportEvent (pts_id uint32, event_type PACKET_KIND, event_data []byte) error {
var err error var err error
switch event_type { switch event_type {
case PACKET_KIND_PEER_STARTED: case PACKET_KIND_PEER_STARTED:
fmt.Printf ("GOT PEER STARTD . CONENCT TO CLIENT_SIDE PEER\n") fmt.Printf("GOT PEER STARTD . CONENCT TO CLIENT_SIDE PEER\n")
r.ptc_wg.Add(1) r.ptc_wg.Add(1)
go r.ConnectToPeer(pts_id, &r.ptc_wg) go r.ConnectToPeer(pts_id, &r.ptc_wg)
case PACKET_KIND_PEER_ABORTED: case PACKET_KIND_PEER_ABORTED:
fallthrough fallthrough
case PACKET_KIND_PEER_STOPPED: case PACKET_KIND_PEER_STOPPED:
fmt.Printf ("GOT PEER STOPPED . DISCONNECTION FROM CLIENT_SIDE PEER\n") fmt.Printf("GOT PEER STOPPED . DISCONNECTION FROM CLIENT_SIDE PEER\n")
err = r.DisconnectFromPeer(pts_id) err = r.DisconnectFromPeer(pts_id)
if err != nil { if err != nil {
// TODO: // TODO:
} }
case PACKET_KIND_PEER_EOF: case PACKET_KIND_PEER_EOF:
fmt.Printf ("GOT PEER EOF. REMEMBER EOF\n") fmt.Printf("GOT PEER EOF. REMEMBER EOF\n")
err = r.CloseWriteToPeer(pts_id) err = r.CloseWriteToPeer(pts_id)
if err != nil { if err != nil {
// TODO: // TODO:
@ -336,8 +337,8 @@ fmt.Printf ("GOT PEER EOF. REMEMBER EOF\n")
} }
// -------------------------------------------------------------------- // --------------------------------------------------------------------
func NewServerConn(c *Client, addr *net.TCPAddr, cfg *ClientConfig) *ServerConn { func NewClientConn(c *Client, addr *net.TCPAddr, cfg *ClientConfig) *ClientConn {
var cts ServerConn var cts ClientConn
cts.cli = c cts.cli = c
cts.route_map = make(ClientRouteMap) cts.route_map = make(ClientRouteMap)
@ -352,37 +353,37 @@ func NewServerConn(c *Client, addr *net.TCPAddr, cfg *ClientConfig) *ServerConn
return &cts return &cts
} }
func (cts *ServerConn) AddNewClientRoute(route_id uint32, addr *net.TCPAddr, proto ROUTE_PROTO) (*ClientRoute, error) { func (cts *ClientConn) AddNewClientRoute(route_id uint32, addr *net.TCPAddr, proto ROUTE_PROTO) (*ClientRoute, error) {
var r *ClientRoute var r *ClientRoute
cts.route_mtx.Lock() cts.route_mtx.Lock()
if cts.route_map[route_id] != nil { if cts.route_map[route_id] != nil {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return nil, fmt.Errorf ("existent route id - %d", route_id) return nil, fmt.Errorf("existent route id - %d", route_id)
} }
r = NewClientRoute(cts, route_id, addr, proto) r = NewClientRoute(cts, route_id, addr, proto)
cts.route_map[route_id] = r cts.route_map[route_id] = r
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
fmt.Printf ("added client route.... %d -> %d\n", route_id, len(cts.route_map)) fmt.Printf("added client route.... %d -> %d\n", route_id, len(cts.route_map))
cts.route_wg.Add(1) cts.route_wg.Add(1)
go r.RunTask(&cts.route_wg) go r.RunTask(&cts.route_wg)
return r, nil return r, nil
} }
func (cts *ServerConn) RemoveClientRoute(route *ClientRoute) error { func (cts *ClientConn) RemoveClientRoute(route *ClientRoute) error {
var r *ClientRoute var r *ClientRoute
var ok bool var ok bool
cts.route_mtx.Lock() cts.route_mtx.Lock()
r, ok = cts.route_map[route.id] r, ok = cts.route_map[route.id]
if (!ok) { if !ok {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route.id) return fmt.Errorf("non-existent route id - %d", route.id)
} }
if r != route { if r != route {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route.id) return fmt.Errorf("non-existent route id - %d", route.id)
} }
delete(cts.route_map, route.id) delete(cts.route_map, route.id)
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
@ -391,15 +392,15 @@ func (cts *ServerConn) RemoveClientRoute(route *ClientRoute) error {
return nil return nil
} }
func (cts *ServerConn) RemoveClientRouteById(route_id uint32) error { func (cts *ClientConn) RemoveClientRouteById(route_id uint32) error {
var r *ClientRoute var r *ClientRoute
var ok bool var ok bool
cts.route_mtx.Lock() cts.route_mtx.Lock()
r, ok = cts.route_map[route_id] r, ok = cts.route_map[route_id]
if (!ok) { if !ok {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route_id) return fmt.Errorf("non-existent route id - %d", route_id)
} }
delete(cts.route_map, route_id) delete(cts.route_map, route_id)
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
@ -408,7 +409,7 @@ func (cts *ServerConn) RemoveClientRouteById(route_id uint32) error {
return nil return nil
} }
func (cts *ServerConn) AddClientRoutes (peer_addrs []string) error { func (cts *ClientConn) AddClientRoutes(peer_addrs []string) error {
var i int var i int
var v string var v string
var addr *net.TCPAddr var addr *net.TCPAddr
@ -436,9 +437,9 @@ func (cts *ServerConn) AddClientRoutes (peer_addrs []string) error {
return nil return nil
} }
func (cts *ServerConn) disconnect_from_server() { func (cts *ClientConn) disconnect_from_server() {
if cts.conn != nil { if cts.conn != nil {
var r* ClientRoute var r *ClientRoute
cts.route_mtx.Lock() cts.route_mtx.Lock()
for _, r = range cts.route_map { for _, r = range cts.route_map {
@ -456,14 +457,14 @@ func (cts *ServerConn) disconnect_from_server() {
} }
} }
func (cts *ServerConn) ReqStop() { func (cts *ClientConn) ReqStop() {
if cts.stop_req.CompareAndSwap(false, true) { if cts.stop_req.CompareAndSwap(false, true) {
cts.disconnect_from_server() cts.disconnect_from_server()
cts.stop_chan <- true cts.stop_chan <- true
} }
} }
func (cts *ServerConn) RunTask(wg *sync.WaitGroup) { func (cts *ClientConn) RunTask(wg *sync.WaitGroup) {
var psc PacketStreamClient var psc PacketStreamClient
var slpctx context.Context var slpctx context.Context
var c_seed Seed var c_seed Seed
@ -473,12 +474,10 @@ func (cts *ServerConn) RunTask(wg *sync.WaitGroup) {
defer wg.Done() // arrange to call at the end of this function defer wg.Done() // arrange to call at the end of this function
start_over: start_over:
cts.cli.log.Write ("", LOG_DEBUG, "Total number of server connections = %d", len(cts.cli.cts_map)) cts.cli.log.Write(cts.lid, LOG_INFO, "Connecting to server %s", cts.saddr.String())
cts.cli.log.Write("", LOG_INFO, "Connecting to server %s", cts.saddr.String())
cts.conn, err = grpc.NewClient(cts.saddr.String(), grpc.WithTransportCredentials(insecure.NewCredentials())) cts.conn, err = grpc.NewClient(cts.saddr.String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil { if err != nil {
cts.cli.log.Write("", LOG_ERROR, "Failed to make client to server %s - %s", cts.saddr.String(), err.Error()) cts.cli.log.Write(cts.lid, LOG_ERROR, "Failed to make client to server %s - %s", cts.saddr.String(), err.Error())
goto reconnect_to_server goto reconnect_to_server
} }
cts.hdc = NewHoduClient(cts.conn) cts.hdc = NewHoduClient(cts.conn)
@ -492,21 +491,21 @@ start_over:
c_seed.Flags = 0 c_seed.Flags = 0
s_seed, err = cts.hdc.GetSeed(cts.cli.ctx, &c_seed) s_seed, err = cts.hdc.GetSeed(cts.cli.ctx, &c_seed)
if err != nil { if err != nil {
cts.cli.log.Write("", LOG_ERROR, "Failed to get seed from server %s - %s", cts.saddr.String(), err.Error()) cts.cli.log.Write(cts.lid, LOG_ERROR, "Failed to get seed from server %s - %s", cts.saddr.String(), err.Error())
goto reconnect_to_server goto reconnect_to_server
} }
cts.s_seed = *s_seed cts.s_seed = *s_seed
cts.c_seed = c_seed cts.c_seed = c_seed
cts.cli.log.Write("", LOG_INFO, "Got seed from server %s - ver=%#x", cts.saddr.String(), cts.s_seed.Version) cts.cli.log.Write(cts.lid, LOG_INFO, "Got seed from server %s - ver=%#x", cts.saddr.String(), cts.s_seed.Version)
psc, err = cts.hdc.PacketStream(cts.cli.ctx) psc, err = cts.hdc.PacketStream(cts.cli.ctx)
if err != nil { if err != nil {
cts.cli.log.Write("", LOG_ERROR, "Failed to get packet stream from server %s - %s", cts.saddr.String(), err.Error()) cts.cli.log.Write(cts.lid, LOG_ERROR, "Failed to get packet stream from server %s - %s", cts.saddr.String(), err.Error())
goto reconnect_to_server goto reconnect_to_server
} }
cts.cli.log.Write("", LOG_INFO, "Got packet stream from server %s", cts.saddr.String()) cts.cli.log.Write(cts.lid, LOG_INFO, "Got packet stream from server %s", cts.saddr.String())
cts.psc = &GuardedPacketStreamClient{Hodu_PacketStreamClient: psc} cts.psc = &GuardedPacketStreamClient{Hodu_PacketStreamClient: psc}
@ -514,7 +513,7 @@ start_over:
// let's add routes to the client-side peers. // let's add routes to the client-side peers.
err = cts.AddClientRoutes(cts.cfg.PeerAddrs) err = cts.AddClientRoutes(cts.cfg.PeerAddrs)
if err != nil { if err != nil {
cts.cli.log.Write("", LOG_INFO, "Failed to add routes to server %s for %v - %s", cts.saddr.String(), cts.cfg.PeerAddrs, err.Error()) cts.cli.log.Write(cts.lid, LOG_INFO, "Failed to add routes to server %s for %v - %s", cts.saddr.String(), cts.cfg.PeerAddrs, err.Error())
goto done goto done
} }
@ -541,7 +540,7 @@ fmt.Printf("context doine... error - %s\n", cts.cli.ctx.Err().Error())
if status.Code(err) == codes.Canceled || errors.Is(err, net.ErrClosed) { if status.Code(err) == codes.Canceled || errors.Is(err, net.ErrClosed) {
goto reconnect_to_server goto reconnect_to_server
} else { } else {
cts.cli.log.Write("", LOG_INFO, "Failed to receive packet form server %s - %s", cts.saddr.String(), err.Error()) cts.cli.log.Write(cts.lid, LOG_INFO, "Failed to receive packet form server %s - %s", cts.saddr.String(), err.Error())
goto reconnect_to_server goto reconnect_to_server
} }
} }
@ -635,7 +634,7 @@ fmt.Printf("context doine... error - %s\n", cts.cli.ctx.Err().Error())
if ok { if ok {
err = cts.ReportEvent(x.Data.RouteId, x.Data.PeerId, PACKET_KIND_PEER_DATA, x.Data.Data) err = cts.ReportEvent(x.Data.RouteId, x.Data.PeerId, PACKET_KIND_PEER_DATA, x.Data.Data)
if err != nil { if err != nil {
fmt.Printf ("failed to report event - %s\n", err.Error()) fmt.Printf("failed to report event - %s\n", err.Error())
// TODO: // TODO:
} else { } else {
// TODO: // TODO:
@ -652,7 +651,7 @@ done:
cts.ReqStop() cts.ReqStop()
wait_for_termination: wait_for_termination:
cts.route_wg.Wait() // wait until all route tasks are finished cts.route_wg.Wait() // wait until all route tasks are finished
cts.cli.RemoveServerConn(cts) cts.cli.RemoveClientConn(cts)
return return
reconnect_to_server: reconnect_to_server:
@ -668,19 +667,19 @@ reconnect_to_server:
// this signal indicates that ReqStop() has been called // this signal indicates that ReqStop() has been called
// so jumt to the waiting label // so jumt to the waiting label
goto wait_for_termination goto wait_for_termination
case <- slpctx.Done(): case <-slpctx.Done():
// do nothing // do nothing
} }
goto start_over // and reconnect goto start_over // and reconnect
} }
func (cts *ServerConn) ReportEvent (route_id uint32, pts_id uint32, event_type PACKET_KIND, event_data []byte) error { func (cts *ClientConn) ReportEvent (route_id uint32, pts_id uint32, event_type PACKET_KIND, event_data []byte) error {
var r *ClientRoute var r *ClientRoute
var ok bool var ok bool
cts.route_mtx.Lock() cts.route_mtx.Lock()
r, ok = cts.route_map[route_id] r, ok = cts.route_map[route_id]
if (!ok) { if !ok {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route_id) return fmt.Errorf ("non-existent route id - %d", route_id)
} }
@ -690,7 +689,7 @@ func (cts *ServerConn) ReportEvent (route_id uint32, pts_id uint32, event_type P
} }
// -------------------------------------------------------------------- // --------------------------------------------------------------------
func (r *ClientRoute) AddNewClientPeerConn (c *net.TCPConn, pts_id uint32) (*ClientPeerConn, error) { func (r *ClientRoute) AddNewClientPeerConn(c *net.TCPConn, pts_id uint32) (*ClientPeerConn, error) {
var ptc *ClientPeerConn var ptc *ClientPeerConn
r.ptc_mtx.Lock() r.ptc_mtx.Lock()
@ -704,31 +703,39 @@ func (r *ClientRoute) AddNewClientPeerConn (c *net.TCPConn, pts_id uint32) (*Cli
// -------------------------------------------------------------------- // --------------------------------------------------------------------
func NewClient(ctx context.Context, listen_on string, logger Logger, tlscfg *tls.Config) *Client { func NewClient(ctx context.Context, listen_on string, logger Logger, tlscfg *tls.Config) *Client {
var c Client var c Client
c.ctx, c.ctx_cancel = context.WithCancel(ctx) c.ctx, c.ctx_cancel = context.WithCancel(ctx)
c.tlscfg = tlscfg c.tlscfg = tlscfg
c.ext_svcs = make([]Service, 0, 1) c.ext_svcs = make([]Service, 0, 1)
c.cts_map = make(ServerConnMap) // TODO: make it configurable... c.cts_map = make(ClientConnMap)
c.cts_map_by_id = make(ClientConnMapById)
c.stop_req.Store(false) c.stop_req.Store(false)
c.stop_chan = make(chan bool, 8) c.stop_chan = make(chan bool, 8)
c.log = logger c.log = logger
c.mux = http.NewServeMux()
c.mux.Handle("/servers", &client_ctl_servers{c: &c})
c.mux.Handle("/servers/{id}", &client_ctl_servers_id{c: &c})
c.mux.Handle("/clients", &client_ctl_clients{c: &c})
c.mux.Handle("/clients/{id}", &client_ctl_clients_id{c: &c})
c.ctl = &http.Server{ c.ctl = &http.Server{
Addr: listen_on, Addr: listen_on,
Handler: &c, Handler: c.mux,
// TODO: more settings
} }
return &c return &c
} }
func (c *Client) AddNewServerConn(addr *net.TCPAddr, cfg *ClientConfig) (*ServerConn, error) { func (c *Client) AddNewClientConn(addr *net.TCPAddr, cfg *ClientConfig) (*ClientConn, error) {
var cts *ServerConn var cts *ClientConn
var ok bool var ok bool
var id uint32
cts = NewServerConn(c, addr, cfg) cts = NewClientConn(c, addr, cfg)
c.cts_mtx.Lock() c.cts_mtx.Lock()
defer c.cts_mtx.Unlock() defer c.cts_mtx.Unlock()
@ -738,23 +745,34 @@ func (c *Client) AddNewServerConn(addr *net.TCPAddr, cfg *ClientConfig) (*Server
return nil, fmt.Errorf("existing server - %s", addr.String()) return nil, fmt.Errorf("existing server - %s", addr.String())
} }
id = rand.Uint32()
for {
_, ok = c.cts_map_by_id[id]
if !ok { break }
id++
}
cts.id = id
cts.lid = fmt.Sprintf("%d", id)
c.cts_map[addr] = cts c.cts_map[addr] = cts
fmt.Printf ("ADD total servers %d\n", len(c.cts_map)) c.cts_map_by_id[id] = cts
fmt.Printf("ADD total servers %d\n", len(c.cts_map))
return cts, nil return cts, nil
} }
func (c *Client) RemoveServerConn(cts *ServerConn) { func (c *Client) RemoveClientConn(cts *ClientConn) {
c.cts_mtx.Lock() c.cts_mtx.Lock()
delete(c.cts_map, cts.saddr) delete(c.cts_map, cts.saddr)
fmt.Printf ("REMOVEDDDDDD CONNECTION FROM %s total servers %d\n", cts.saddr, len(c.cts_map)) delete(c.cts_map_by_id, cts.id)
fmt.Printf("REMOVEDDDDDD CONNECTION FROM %s total servers %d\n", cts.saddr, len(c.cts_map))
c.cts_mtx.Unlock() c.cts_mtx.Unlock()
} }
func (c *Client) ReqStop() { func (c *Client) ReqStop() {
if c.stop_req.CompareAndSwap(false, true) { if c.stop_req.CompareAndSwap(false, true) {
var cts *ServerConn var cts *ClientConn
if (c.ctl != nil) { if c.ctl != nil {
c.ctl.Shutdown(c.ctx) // to break c.ctl.ListenAndServe() c.ctl.Shutdown(c.ctx) // to break c.ctl.ListenAndServe()
} }
@ -765,67 +783,18 @@ func (c *Client) ReqStop() {
c.stop_chan <- true c.stop_chan <- true
c.ctx_cancel() c.ctx_cancel()
} }
fmt.Printf ("*** Sent stop request to client..\n")
} }
func (c *Client) ServeHTTP(w http.ResponseWriter, req *http.Request) {
var err error
// command handler for the control channel
if req.URL.String() == "/servers" {
switch req.Method {
case http.MethodGet:
goto bad_request // TODO:
case http.MethodPost:
var s ClientCtlParamServer
var cc ClientConfig
err = json.NewDecoder(req.Body).Decode(&s)
if err != nil {
fmt.Printf ("failed to decode body - %s\n", err.Error())
goto bad_request
}
cc.ServerAddr = s.ServerAddr
cc.PeerAddrs = s.PeerAddrs
c.StartService(&cc) // TODO: this can be blocking. do we have to resolve addresses before calling this? also not good because resolution succeed or fail at each attempt. however ok as ServeHTTP itself is in a goroutine?
w.WriteHeader(http.StatusCreated)
case http.MethodPut:
goto bad_request // TODO:
case http.MethodDelete:
var cts *ServerConn
for _, cts = range c.cts_map {
cts.ReqStop()
}
}
} else {
goto bad_request
}
fmt.Printf ("[%s][%s][%s]\n", req.RequestURI, req.URL.String(), req.Method)
return
bad_request:
w.WriteHeader(http.StatusBadRequest)
return
}
/*
* POST GET PUT DELETE
* /servers - create new server list all servers bulk update delete all servers
* /servers/1 - X get server 1 details update server 1 delete server 1
* /servers/1/xxx -
*/
func (c *Client) RunCtlTask(wg *sync.WaitGroup) { func (c *Client) RunCtlTask(wg *sync.WaitGroup) {
var err error var err error
defer wg.Done() defer wg.Done()
err = c.ctl.ListenAndServe() err = c.ctl.ListenAndServe()
if !errors.Is(err, http.ErrServerClosed) { if errors.Is(err, http.ErrServerClosed) {
fmt.Printf ("------------http server error - %s\n", err.Error()) c.log.Write("", LOG_DEBUG, "Control channel closed")
} else { } else {
fmt.Printf ("********* http server ended\n") c.log.Write("", LOG_ERROR, "Control channel error - %s", err.Error())
} }
} }
@ -845,7 +814,7 @@ func (c *Client) RunTask(wg *sync.WaitGroup) {
// RunTask - supposed to be detached as a go routine // RunTask - supposed to be detached as a go routine
func (c *Client) StartService(data interface{}) { func (c *Client) StartService(data interface{}) {
var saddr *net.TCPAddr var saddr *net.TCPAddr
var cts *ServerConn var cts *ClientConn
var err error var err error
var cfg *ClientConfig var cfg *ClientConfig
var ok bool var ok bool
@ -867,7 +836,7 @@ func (c *Client) StartService(data interface{}) {
return return
} }
cts, err = c.AddNewServerConn(saddr, cfg) cts, err = c.AddNewClientConn(saddr, cfg)
if err != nil { if err != nil {
fmt.Printf("unable to add server connection structure to %s - %s", cfg.ServerAddr, err.Error()) fmt.Printf("unable to add server connection structure to %s - %s", cfg.ServerAddr, err.Error())
return return
@ -895,6 +864,6 @@ func (c *Client) WaitForTermination() {
c.wg.Wait() c.wg.Wait()
} }
func (c *Client) WriteLog (id string, level LogLevel, fmtstr string, args ...interface{}) { func (c *Client) WriteLog(id string, level LogLevel, fmtstr string, args ...interface{}) {
c.log.Write(id, level, fmtstr, args...) c.log.Write(id, level, fmtstr, args...)
} }

2
go.mod
View File

@ -1,6 +1,6 @@
module hodu module hodu
go 1.21.0 go 1.22.0
require ( require (
github.com/google/uuid v1.6.0 github.com/google/uuid v1.6.0

14
packet.go
View File

@ -1,6 +1,5 @@
package hodu package hodu
func MakeRouteStartPacket(route_id uint32, proto ROUTE_PROTO, addr string) *Packet { func MakeRouteStartPacket(route_id uint32, proto ROUTE_PROTO, addr string) *Packet {
return &Packet{ return &Packet{
Kind: PACKET_KIND_ROUTE_START, Kind: PACKET_KIND_ROUTE_START,
@ -25,7 +24,6 @@ func MakeRouteStoppedPacket(route_id uint32, proto ROUTE_PROTO) *Packet {
U: &Packet_Route{Route: &RouteDesc{RouteId: route_id, Proto: proto}}} U: &Packet_Route{Route: &RouteDesc{RouteId: route_id, Proto: proto}}}
} }
func MakePeerStartedPacket(route_id uint32, peer_id uint32) *Packet { func MakePeerStartedPacket(route_id uint32, peer_id uint32) *Packet {
// the connection from a peer to the server has been established // the connection from a peer to the server has been established
return &Packet{Kind: PACKET_KIND_PEER_STARTED, return &Packet{Kind: PACKET_KIND_PEER_STARTED,
@ -35,24 +33,20 @@ func MakePeerStartedPacket(route_id uint32, peer_id uint32) *Packet {
func MakePeerStoppedPacket(route_id uint32, peer_id uint32) *Packet { func MakePeerStoppedPacket(route_id uint32, peer_id uint32) *Packet {
return &Packet{Kind: PACKET_KIND_PEER_STOPPED, return &Packet{Kind: PACKET_KIND_PEER_STOPPED,
U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}, U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}}}
}}
} }
func MakePeerAbortedPacket(route_id uint32, peer_id uint32) *Packet { func MakePeerAbortedPacket(route_id uint32, peer_id uint32) *Packet {
return &Packet{Kind: PACKET_KIND_PEER_ABORTED, return &Packet{Kind: PACKET_KIND_PEER_ABORTED,
U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}, U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}}}
}}
} }
func MakePeerEofPacket(route_id uint32, peer_id uint32) *Packet { func MakePeerEofPacket(route_id uint32, peer_id uint32) *Packet {
return &Packet{Kind: PACKET_KIND_PEER_EOF, return &Packet{Kind: PACKET_KIND_PEER_EOF,
U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}, U: &Packet_Peer{Peer: &PeerDesc{RouteId: route_id, PeerId: peer_id}}}
}}
} }
func MakePeerDataPacket(route_id uint32, peer_id uint32, data []byte) *Packet { func MakePeerDataPacket(route_id uint32, peer_id uint32, data []byte) *Packet {
return &Packet{Kind: PACKET_KIND_PEER_DATA, return &Packet{Kind: PACKET_KIND_PEER_DATA,
U: &Packet_Data{Data: &PeerData{RouteId: route_id, PeerId: peer_id, Data: data}, U: &Packet_Data{Data: &PeerData{RouteId: route_id, PeerId: peer_id, Data: data}}}
}}
} }

14
s-peer.go → server-peer.go
View File

@ -23,7 +23,7 @@ type ServerPeerConn struct {
client_peer_eof atomic.Bool client_peer_eof atomic.Bool
} }
func NewServerPeerConn(r *ServerRoute, c *net.TCPConn, id uint32) (*ServerPeerConn) { func NewServerPeerConn(r *ServerRoute, c *net.TCPConn, id uint32) *ServerPeerConn {
var spc ServerPeerConn var spc ServerPeerConn
spc.route = r spc.route = r
@ -37,7 +37,7 @@ func NewServerPeerConn(r *ServerRoute, c *net.TCPConn, id uint32) (*ServerPeerCo
spc.client_peer_started.Store(false) spc.client_peer_started.Store(false)
spc.client_peer_stopped.Store(false) spc.client_peer_stopped.Store(false)
spc.client_peer_eof.Store(false) spc.client_peer_eof.Store(false)
fmt.Printf ("~~~~~~~~~~~~~~~ NEW SERVER PEER CONNECTION ADDED %p\n", &spc) fmt.Printf("~~~~~~~~~~~~~~~ NEW SERVER PEER CONNECTION ADDED %p\n", &spc)
return &spc return &spc
} }
@ -76,7 +76,7 @@ wait_for_started:
tmr.Stop() tmr.Stop()
goto done goto done
case <- spc.stop_chan: case <-spc.stop_chan:
tmr.Stop() tmr.Stop()
goto done goto done
} }
@ -110,9 +110,9 @@ wait_for_stopped:
for { for {
fmt.Printf ("******************* Waiting for peer Stop\n") fmt.Printf ("******************* Waiting for peer Stop\n")
select { select {
case status = <- spc.client_peer_status_chan: // something not right... may use a different channel for closing... case status = <-spc.client_peer_status_chan: // something not right... may use a different channel for closing...
goto done goto done
case <- spc.stop_chan: case <-spc.stop_chan:
goto done goto done
} }
} }
@ -149,7 +149,7 @@ func (spc *ServerPeerConn) ReqStop() {
} }
} }
func (spc *ServerPeerConn) ReportEvent (event_type PACKET_KIND, event_data []byte) error { func (spc *ServerPeerConn) ReportEvent(event_type PACKET_KIND, event_data []byte) error {
switch event_type { switch event_type {
case PACKET_KIND_PEER_STARTED: case PACKET_KIND_PEER_STARTED:
@ -184,7 +184,7 @@ fmt.Printf("******************* CCCCCCCCCCCCCCCCCCCCCCCccc\n")
} }
} else { } else {
// protocol error. the client must not relay more data from the client-side peer after EOF. // protocol error. the client must not relay more data from the client-side peer after EOF.
fmt.Printf ("WARNING - broken client - redundant data from %s to %s\n", spc.route.cts.caddr, spc.conn.RemoteAddr().String()) fmt.Printf("WARNING - broken client - redundant data from %s to %s\n", spc.route.cts.caddr, spc.conn.RemoteAddr().String())
} }
default: default:

124
server.go
View File

@ -19,7 +19,7 @@ import "google.golang.org/grpc/stats"
const PTS_LIMIT = 8192 const PTS_LIMIT = 8192
type ClientConnMap = map[net.Addr]*ClientConn type ServerConnMap = map[net.Addr]*ServerConn
type ServerPeerConnMap = map[uint32]*ServerPeerConn type ServerPeerConnMap = map[uint32]*ServerPeerConn
type ServerRouteMap = map[uint32]*ServerRoute type ServerRouteMap = map[uint32]*ServerRoute
@ -39,7 +39,7 @@ type Server struct {
l_wg sync.WaitGroup l_wg sync.WaitGroup
cts_mtx sync.Mutex cts_mtx sync.Mutex
cts_map ClientConnMap cts_map ServerConnMap
cts_wg sync.WaitGroup cts_wg sync.WaitGroup
gs *grpc.Server gs *grpc.Server
@ -48,9 +48,9 @@ type Server struct {
UnimplementedHoduServer UnimplementedHoduServer
} }
// client connection to server. // connection from client.
// client connect to the server, the server accept it, and makes a tunnel request // client connect to the server, the server accept it, and makes a tunnel request
type ClientConn struct { type ServerConn struct {
svr *Server svr *Server
caddr net.Addr // client address that created this structure caddr net.Addr // client address that created this structure
pss *GuardedPacketStreamServer pss *GuardedPacketStreamServer
@ -65,7 +65,7 @@ type ClientConn struct {
} }
type ServerRoute struct { type ServerRoute struct {
cts *ClientConn cts *ServerConn
l *net.TCPListener l *net.TCPListener
laddr *net.TCPAddr laddr *net.TCPAddr
id uint32 id uint32
@ -107,7 +107,7 @@ func (g *GuardedPacketStreamServer) Context() context.Context {
// ------------------------------------ // ------------------------------------
func NewServerRoute(cts *ClientConn, id uint32, proto ROUTE_PROTO) (*ServerRoute, error) { func NewServerRoute(cts *ServerConn, id uint32, proto ROUTE_PROTO) (*ServerRoute, error) {
var r ServerRoute var r ServerRoute
var l *net.TCPListener var l *net.TCPListener
var laddr *net.TCPAddr var laddr *net.TCPAddr
@ -209,7 +209,7 @@ func (r *ServerRoute) RunTask(wg *sync.WaitGroup) {
} }
func (r *ServerRoute) ReqStop() { func (r *ServerRoute) ReqStop() {
fmt.Printf ("requesting to stop route taak..\n") fmt.Printf("requesting to stop route taak..\n")
if r.stop_req.CompareAndSwap(false, true) { if r.stop_req.CompareAndSwap(false, true) {
var pts *ServerPeerConn var pts *ServerPeerConn
@ -220,10 +220,10 @@ func (r *ServerRoute) ReqStop() {
r.l.Close() r.l.Close()
} }
fmt.Printf ("requiested to stopp route taak..\n") fmt.Printf("requiested to stopp route taak..\n")
} }
func (r *ServerRoute) ReportEvent (pts_id uint32, event_type PACKET_KIND, event_data []byte) error { func (r *ServerRoute) ReportEvent(pts_id uint32, event_type PACKET_KIND, event_data []byte) error {
var spc *ServerPeerConn var spc *ServerPeerConn
var ok bool var ok bool
@ -239,7 +239,7 @@ func (r *ServerRoute) ReportEvent (pts_id uint32, event_type PACKET_KIND, event_
} }
// ------------------------------------ // ------------------------------------
func (cts *ClientConn) make_route_listener(proto ROUTE_PROTO) (*net.TCPListener, *net.TCPAddr, error) { func (cts *ServerConn) make_route_listener(proto ROUTE_PROTO) (*net.TCPListener, *net.TCPAddr, error) {
var l *net.TCPListener var l *net.TCPListener
var err error var err error
var laddr *net.TCPAddr var laddr *net.TCPAddr
@ -279,14 +279,14 @@ func (cts *ClientConn) make_route_listener(proto ROUTE_PROTO) (*net.TCPListener,
return nil, nil, err return nil, nil, err
} }
func (cts *ClientConn) AddNewServerRoute(route_id uint32, proto ROUTE_PROTO) (*ServerRoute, error) { func (cts *ServerConn) AddNewServerRoute(route_id uint32, proto ROUTE_PROTO) (*ServerRoute, error) {
var r *ServerRoute var r *ServerRoute
var err error var err error
cts.route_mtx.Lock() cts.route_mtx.Lock()
if cts.route_map[route_id] != nil { if cts.route_map[route_id] != nil {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return nil, fmt.Errorf ("existent route id - %d", route_id) return nil, fmt.Errorf("existent route id - %d", route_id)
} }
r, err = NewServerRoute(cts, route_id, proto) r, err = NewServerRoute(cts, route_id, proto)
if err != nil { if err != nil {
@ -301,19 +301,19 @@ func (cts *ClientConn) AddNewServerRoute(route_id uint32, proto ROUTE_PROTO) (*S
return r, nil return r, nil
} }
func (cts *ClientConn) RemoveServerRoute (route* ServerRoute) error { func (cts *ServerConn) RemoveServerRoute(route *ServerRoute) error {
var r *ServerRoute var r *ServerRoute
var ok bool var ok bool
cts.route_mtx.Lock() cts.route_mtx.Lock()
r, ok = cts.route_map[route.id] r, ok = cts.route_map[route.id]
if (!ok) { if !ok {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route.id) return fmt.Errorf("non-existent route id - %d", route.id)
} }
if (r != route) { if r != route {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route - %d", route.id) return fmt.Errorf("non-existent route - %d", route.id)
} }
delete(cts.route_map, route.id) delete(cts.route_map, route.id)
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
@ -322,15 +322,15 @@ func (cts *ClientConn) RemoveServerRoute (route* ServerRoute) error {
return nil return nil
} }
func (cts *ClientConn) RemoveServerRouteById (route_id uint32) (*ServerRoute, error) { func (cts *ServerConn) RemoveServerRouteById(route_id uint32) (*ServerRoute, error) {
var r *ServerRoute var r *ServerRoute
var ok bool var ok bool
cts.route_mtx.Lock() cts.route_mtx.Lock()
r, ok = cts.route_map[route_id] r, ok = cts.route_map[route_id]
if (!ok) { if !ok {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return nil, fmt.Errorf ("non-existent route id - %d", route_id) return nil, fmt.Errorf("non-existent route id - %d", route_id)
} }
delete(cts.route_map, route_id) delete(cts.route_map, route_id)
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
@ -339,7 +339,7 @@ func (cts *ClientConn) RemoveServerRouteById (route_id uint32) (*ServerRoute, er
return r, nil return r, nil
} }
func (cts *ClientConn) ReportEvent (route_id uint32, pts_id uint32, event_type PACKET_KIND, event_data []byte) error { func (cts *ServerConn) ReportEvent(route_id uint32, pts_id uint32, event_type PACKET_KIND, event_data []byte) error {
var r *ServerRoute var r *ServerRoute
var ok bool var ok bool
@ -347,14 +347,14 @@ func (cts *ClientConn) ReportEvent (route_id uint32, pts_id uint32, event_type P
r, ok = cts.route_map[route_id] r, ok = cts.route_map[route_id]
if (!ok) { if (!ok) {
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return fmt.Errorf ("non-existent route id - %d", route_id) return fmt.Errorf("non-existent route id - %d", route_id)
} }
cts.route_mtx.Unlock() cts.route_mtx.Unlock()
return r.ReportEvent(pts_id, event_type, event_data) return r.ReportEvent(pts_id, event_type, event_data)
} }
func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) { func (cts *ServerConn) receive_from_stream(wg *sync.WaitGroup) {
var pkt *Packet var pkt *Packet
var err error var err error
@ -377,7 +377,7 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
var ok bool var ok bool
x, ok = pkt.U.(*Packet_Route) x, ok = pkt.U.(*Packet_Route)
if ok { if ok {
var r* ServerRoute var r *ServerRoute
r, err = cts.AddNewServerRoute(x.Route.RouteId, x.Route.Proto) r, err = cts.AddNewServerRoute(x.Route.RouteId, x.Route.Proto)
if err != nil { if err != nil {
@ -401,7 +401,7 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
var ok bool var ok bool
x, ok = pkt.U.(*Packet_Route) x, ok = pkt.U.(*Packet_Route)
if ok { if ok {
var r* ServerRoute var r *ServerRoute
r, err = cts.RemoveServerRouteById(x.Route.RouteId) r, err = cts.RemoveServerRouteById(x.Route.RouteId)
if err != nil { if err != nil {
@ -429,7 +429,7 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
err = cts.ReportEvent(x.Peer.RouteId, x.Peer.PeerId, PACKET_KIND_PEER_STARTED, nil) err = cts.ReportEvent(x.Peer.RouteId, x.Peer.PeerId, PACKET_KIND_PEER_STARTED, nil)
if err != nil { if err != nil {
// TODO: // TODO:
fmt.Printf ("Failed to report PEER_STARTED Event") fmt.Printf("Failed to report PEER_STARTED Event")
} else { } else {
// TODO: // TODO:
} }
@ -448,7 +448,7 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
err = cts.ReportEvent(x.Peer.RouteId, x.Peer.PeerId, PACKET_KIND_PEER_STOPPED, nil) err = cts.ReportEvent(x.Peer.RouteId, x.Peer.PeerId, PACKET_KIND_PEER_STOPPED, nil)
if err != nil { if err != nil {
// TODO: // TODO:
fmt.Printf ("Failed to report PEER_STOPPED Event") fmt.Printf("Failed to report PEER_STOPPED Event")
} else { } else {
// TODO: // TODO:
} }
@ -465,7 +465,7 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
err = cts.ReportEvent(x.Data.RouteId, x.Data.PeerId, PACKET_KIND_PEER_DATA, x.Data.Data) err = cts.ReportEvent(x.Data.RouteId, x.Data.PeerId, PACKET_KIND_PEER_DATA, x.Data.Data)
if err != nil { if err != nil {
// TODO: // TODO:
fmt.Printf ("Failed to report PEER_DATA Event") fmt.Printf("Failed to report PEER_DATA Event")
} else { } else {
// TODO: // TODO:
} }
@ -476,10 +476,10 @@ func (cts *ClientConn) receive_from_stream(wg *sync.WaitGroup) {
} }
done: done:
fmt.Printf ("************ stream receiver finished....\n") fmt.Printf("************ stream receiver finished....\n")
} }
func (cts *ClientConn) RunTask(wg *sync.WaitGroup) { func (cts *ServerConn) RunTask(wg *sync.WaitGroup) {
var strm *GuardedPacketStreamServer var strm *GuardedPacketStreamServer
var ctx context.Context var ctx context.Context
@ -520,13 +520,13 @@ fmt.Printf("grpc server done - %s\n", ctx.Err().Error())
} }
done: done:
fmt.Printf ("^^^^^^^^^^^^^^^^^ waiting for reoute_wg...\n") fmt.Printf("^^^^^^^^^^^^^^^^^ waiting for reoute_wg...\n")
cts.ReqStop() // just in case cts.ReqStop() // just in case
cts.route_wg.Wait() cts.route_wg.Wait()
fmt.Printf ("^^^^^^^^^^^^^^^^^ waited for reoute_wg...\n") fmt.Printf("^^^^^^^^^^^^^^^^^ waited for reoute_wg...\n")
} }
func (cts *ClientConn) ReqStop() { func (cts *ServerConn) ReqStop() {
if cts.stop_req.CompareAndSwap(false, true) { if cts.stop_req.CompareAndSwap(false, true) {
var r *ServerRoute var r *ServerRoute
@ -538,14 +538,14 @@ func (cts *ClientConn) ReqStop() {
// the grpc server. while the global grpc server is closed in // the grpc server. while the global grpc server is closed in
// ReqStop() for Server, the individuation connection is closed // ReqStop() for Server, the individuation connection is closed
// by returing from the grpc handler goroutine. See the comment // by returing from the grpc handler goroutine. See the comment
// RunTask() for ClientConn. // RunTask() for ServerConn.
cts.stop_chan <- true cts.stop_chan <- true
} }
} }
// -------------------------------------------------------------------- // --------------------------------------------------------------------
func (s *Server) GetSeed (ctx context.Context, c_seed *Seed) (*Seed, error) { func (s *Server) GetSeed(ctx context.Context, c_seed *Seed) (*Seed, error) {
var s_seed Seed var s_seed Seed
// seed exchange is for furture expansion of the protocol // seed exchange is for furture expansion of the protocol
@ -554,7 +554,7 @@ func (s *Server) GetSeed (ctx context.Context, c_seed *Seed) (*Seed, error) {
s_seed.Version = HODU_VERSION s_seed.Version = HODU_VERSION
s_seed.Flags = 0 s_seed.Flags = 0
// we create no ClientConn structure associated with the connection // we create no ServerConn structure associated with the connection
// at this phase for the server. it doesn't track the client version and // at this phase for the server. it doesn't track the client version and
// features. we delegate protocol selection solely to the client. // features. we delegate protocol selection solely to the client.
@ -566,15 +566,15 @@ func (s *Server) PacketStream(strm Hodu_PacketStreamServer) error {
var p *peer.Peer var p *peer.Peer
var ok bool var ok bool
var err error var err error
var cts *ClientConn var cts *ServerConn
ctx = strm.Context() ctx = strm.Context()
p, ok = peer.FromContext(ctx) p, ok = peer.FromContext(ctx)
if (!ok) { if !ok {
return fmt.Errorf("failed to get peer from packet stream context") return fmt.Errorf("failed to get peer from packet stream context")
} }
cts, err = s.AddNewClientConn(p.Addr, strm) cts, err = s.AddNewServerConn(p.Addr, strm)
if err != nil { if err != nil {
return fmt.Errorf("unable to add client %s - %s", p.Addr.String(), err.Error()) return fmt.Errorf("unable to add client %s - %s", p.Addr.String(), err.Error())
} }
@ -593,7 +593,7 @@ type ConnCatcher struct {
} }
func (cc *ConnCatcher) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { func (cc *ConnCatcher) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
return ctx return ctx
} }
func (cc *ConnCatcher) HandleRPC(ctx context.Context, s stats.RPCStats) { func (cc *ConnCatcher) HandleRPC(ctx context.Context, s stats.RPCStats) {
@ -601,7 +601,7 @@ func (cc *ConnCatcher) HandleRPC(ctx context.Context, s stats.RPCStats) {
func (cc *ConnCatcher) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context { func (cc *ConnCatcher) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
return ctx return ctx
//return context.TODO() //return context.TODO()
} }
func (cc *ConnCatcher) HandleConn(ctx context.Context, cs stats.ConnStats) { func (cc *ConnCatcher) HandleConn(ctx context.Context, cs stats.ConnStats) {
@ -611,7 +611,7 @@ func (cc *ConnCatcher) HandleConn(ctx context.Context, cs stats.ConnStats) {
var addr string var addr string
p, ok = peer.FromContext(ctx) p, ok = peer.FromContext(ctx)
if (!ok) { if !ok {
addr = "" addr = ""
} else { } else {
addr = p.Addr.String() addr = p.Addr.String()
@ -626,8 +626,8 @@ if ok {
fmt.Printf("**** client connected - [%s]\n", addr) fmt.Printf("**** client connected - [%s]\n", addr)
case *stats.ConnEnd: case *stats.ConnEnd:
fmt.Printf("**** client disconnected - [%s]\n", addr) fmt.Printf("**** client disconnected - [%s]\n", addr)
cc.server.RemoveClientConnByAddr(p.Addr) cc.server.RemoveServerConnByAddr(p.Addr)
} }
} }
// wrappedStream wraps around the embedded grpc.ServerStream, and intercepts the RecvMsg and // wrappedStream wraps around the embedded grpc.ServerStream, and intercepts the RecvMsg and
@ -720,7 +720,7 @@ func NewServer(ctx context.Context, laddrs []string, logger Logger, tlscfg *tls.
s.tlscfg = tlscfg s.tlscfg = tlscfg
s.ext_svcs = make([]Service, 0, 1) s.ext_svcs = make([]Service, 0, 1)
s.cts_map = make(ClientConnMap) // TODO: make it configurable... s.cts_map = make(ServerConnMap) // TODO: make it configurable...
s.stop_chan = make(chan bool, 8) s.stop_chan = make(chan bool, 8)
s.stop_req.Store(false) s.stop_req.Store(false)
/* /*
@ -733,9 +733,9 @@ func NewServer(ctx context.Context, laddrs []string, logger Logger, tlscfg *tls.
s.gs = grpc.NewServer( s.gs = grpc.NewServer(
grpc.UnaryInterceptor(unaryInterceptor), grpc.UnaryInterceptor(unaryInterceptor),
grpc.StreamInterceptor(streamInterceptor), grpc.StreamInterceptor(streamInterceptor),
grpc.StatsHandler(&ConnCatcher{ server: &s }), grpc.StatsHandler(&ConnCatcher{server: &s}),
) // TODO: have this outside the server struct? ) // TODO: have this outside the server struct?
RegisterHoduServer (s.gs, &s) RegisterHoduServer(s.gs, &s)
return &s, nil return &s, nil
@ -760,13 +760,13 @@ func (s *Server) run_grpc_server(idx int, wg *sync.WaitGroup) error {
l = s.l[idx] l = s.l[idx]
// it seems to be safe to call a single grpc server on differnt listening sockets multiple times // it seems to be safe to call a single grpc server on differnt listening sockets multiple times
s.log.Write ("", LOG_ERROR, "Starting GRPC server listening on %s", l.Addr().String()) s.log.Write("", LOG_ERROR, "Starting GRPC server listening on %s", l.Addr().String())
err = s.gs.Serve(l) err = s.gs.Serve(l)
if err != nil { if err != nil {
if errors.Is(err, net.ErrClosed) { if errors.Is(err, net.ErrClosed) {
s.log.Write ("", LOG_ERROR, "GRPC server listening on %s closed", l.Addr().String()) s.log.Write("", LOG_ERROR, "GRPC server listening on %s closed", l.Addr().String())
} else { } else {
s.log.Write ("", LOG_ERROR, "Error from GRPC server listening on %s - %s", l.Addr().String(), err.Error()) s.log.Write("", LOG_ERROR, "Error from GRPC server listening on %s - %s", l.Addr().String(), err.Error())
} }
return err return err
} }
@ -814,18 +814,18 @@ func (s *Server) RunCtlTask(wg *sync.WaitGroup) {
err = s.ctl.ListenAndServe() err = s.ctl.ListenAndServe()
if errors.Is(err, http.ErrServerClosed) { if errors.Is(err, http.ErrServerClosed) {
fmt.Printf ("------------http server error - %s\n", err.Error()) fmt.Printf("------------http server error - %s\n", err.Error())
} else { } else {
fmt.Printf ("********* http server ended\n") fmt.Printf("********* http server ended\n")
} }
} }
func (s *Server) ReqStop() { func (s *Server) ReqStop() {
if s.stop_req.CompareAndSwap(false, true) { if s.stop_req.CompareAndSwap(false, true) {
var l *net.TCPListener var l *net.TCPListener
var cts *ClientConn var cts *ServerConn
if (s.ctl != nil) { if s.ctl != nil {
// shutdown the control server if ever started. // shutdown the control server if ever started.
s.ctl.Shutdown(s.ctx) s.ctl.Shutdown(s.ctx)
} }
@ -847,8 +847,8 @@ func (s *Server) ReqStop() {
} }
} }
func (s *Server) AddNewClientConn(addr net.Addr, pss Hodu_PacketStreamServer) (*ClientConn, error) { func (s *Server) AddNewServerConn(addr net.Addr, pss Hodu_PacketStreamServer) (*ServerConn, error) {
var cts ClientConn var cts ServerConn
var ok bool var ok bool
cts.svr = s cts.svr = s
@ -871,15 +871,15 @@ func (s *Server) AddNewClientConn(addr net.Addr, pss Hodu_PacketStreamServer) (*
return &cts, nil return &cts, nil
} }
func (s *Server) RemoveClientConn(cts *ClientConn) { func (s *Server) RemoveServerConn(cts *ServerConn) {
s.cts_mtx.Lock() s.cts_mtx.Lock()
delete(s.cts_map, cts.caddr) delete(s.cts_map, cts.caddr)
s.log.Write("", LOG_DEBUG, "Removed client connection from %s", cts.caddr.String()) s.log.Write("", LOG_DEBUG, "Removed client connection from %s", cts.caddr.String())
s.cts_mtx.Unlock() s.cts_mtx.Unlock()
} }
func (s *Server) RemoveClientConnByAddr(addr net.Addr) { func (s *Server) RemoveServerConnByAddr(addr net.Addr) {
var cts *ClientConn var cts *ServerConn
var ok bool var ok bool
s.cts_mtx.Lock() s.cts_mtx.Lock()
@ -892,8 +892,8 @@ func (s *Server) RemoveClientConnByAddr(addr net.Addr) {
} }
} }
func (s *Server) FindClientConnByAddr (addr net.Addr) *ClientConn { func (s *Server) FindServerConnByAddr(addr net.Addr) *ServerConn {
var cts *ClientConn var cts *ServerConn
var ok bool var ok bool
s.cts_mtx.Lock() s.cts_mtx.Lock()
@ -930,6 +930,6 @@ func (s *Server) WaitForTermination() {
s.wg.Wait() s.wg.Wait()
} }
func (s *Server) WriteLog (id string, level LogLevel, fmtstr string, args ...interface{}) { func (s *Server) WriteLog(id string, level LogLevel, fmtstr string, args ...interface{}) {
s.log.Write(id, level, fmtstr, args...) s.log.Write(id, level, fmtstr, args...)
} }