unistack-org/micro
4
0
Fork 3
Code Issues 20 Pull Requests 2 Actions Packages Projects Releases 284 Wiki Activity

Merge pull request #786 from micro/net

Browse Source 
Do not embed proxy/router/tunnel
This commit is contained in:
Asim Aslam 2019-09-25 13:00:36 +01:00 committed by GitHub
commit 855cd5ecf4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 22 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
network/default.go
View File

@ -42,11 +42,11 @@ type network struct {
// options configure the network // options configure the network
options Options options Options
// rtr is network router // rtr is network router
router.Router router router.Router
// prx is network proxy // prx is network proxy
proxy.Proxy proxy proxy.Proxy
// tun is network tunnel // tun is network tunnel
tunnel.Tunnel tunnel tunnel.Tunnel
// server is network server // server is network server
server server.Server server server.Server
// client is network client // client is network client
@ -118,9 +118,9 @@ func newNetwork(opts ...Option) Network {
peers: make(map[string]*node), peers: make(map[string]*node),
}, },
options: options, options: options,
Router: options.Router, router: options.Router,
Proxy: options.Proxy, proxy: options.Proxy,
Tunnel: options.Tunnel, tunnel: options.Tunnel,
server: server, server: server,
client: client, client: client,
tunClient: make(map[string]transport.Client), tunClient: make(map[string]transport.Client),
@ -146,11 +146,6 @@ func (n *network) Name() string {
return n.options.Name return n.options.Name
} }
// Address returns network bind address
func (n *network) Address() string {
return n.Tunnel.Address()
}
// resolveNodes resolves network nodes to addresses // resolveNodes resolves network nodes to addresses
func (n *network) resolveNodes() ([]string, error) { func (n *network) resolveNodes() ([]string, error) {
// resolve the network address to network nodes // resolve the network address to network nodes
@ -194,7 +189,7 @@ func (n *network) resolve() {
continue continue
} }
// initialize the tunnel // initialize the tunnel
n.Tunnel.Init( n.tunnel.Init(
tunnel.Nodes(nodes...), tunnel.Nodes(nodes...),
) )
} }
@ -286,7 +281,7 @@ func (n *network) processNetChan(client transport.Client, listener tunnel.Listen
log.Debugf("Network failed to advertise peers: %v", err) log.Debugf("Network failed to advertise peers: %v", err)
} }
// advertise all the routes when a new node has connected // advertise all the routes when a new node has connected
if err := n.Router.Solicit(); err != nil { if err := n.router.Solicit(); err != nil {
log.Debugf("Network failed to solicit routes: %s", err) log.Debugf("Network failed to solicit routes: %s", err)
} }
case "peer": case "peer":
@ -410,13 +405,13 @@ func (n *network) announce(client transport.Client) {
// pruneRoutes prunes routes return by given query // pruneRoutes prunes routes return by given query
func (n *network) pruneRoutes(q router.Query) error { func (n *network) pruneRoutes(q router.Query) error {
routes, err := n.Router.Table().Query(q) routes, err := n.router.Table().Query(q)
if err != nil && err != router.ErrRouteNotFound { if err != nil && err != router.ErrRouteNotFound {
return err return err
} }
for _, route := range routes { for _, route := range routes {
if err := n.Router.Table().Delete(route); err != nil && err != router.ErrRouteNotFound { if err := n.router.Table().Delete(route); err != nil && err != router.ErrRouteNotFound {
return err return err
} }
} }
@ -622,7 +617,7 @@ func (n *network) processCtrlChan(client transport.Client, listener tunnel.Liste
} }
log.Debugf("Network router processing advert: %s", advert.Id) log.Debugf("Network router processing advert: %s", advert.Id)
if err := n.Router.Process(advert); err != nil { if err := n.router.Process(advert); err != nil {
log.Debugf("Network failed to process advert %s: %v", advert.Id, err) log.Debugf("Network failed to process advert %s: %v", advert.Id, err)
} }
case "solicit": case "solicit":
@ -638,7 +633,7 @@ func (n *network) processCtrlChan(client transport.Client, listener tunnel.Liste
} }
log.Debugf("Network router flushing routes for: %s", pbRtrSolicit.Id) log.Debugf("Network router flushing routes for: %s", pbRtrSolicit.Id)
// advertise all the routes when a new node has connected // advertise all the routes when a new node has connected
if err := n.Router.Solicit(); err != nil { if err := n.router.Solicit(); err != nil {
log.Debugf("Network failed to solicit routes: %s", err) log.Debugf("Network failed to solicit routes: %s", err)
} }
} }
@ -719,7 +714,7 @@ func (n *network) Connect() error {
} }
// connect network tunnel // connect network tunnel
if err := n.Tunnel.Connect(); err != nil { if err := n.tunnel.Connect(); err != nil {
n.Unlock() n.Unlock()
return err return err
} }
@ -727,17 +722,17 @@ func (n *network) Connect() error {
// set our internal node address // set our internal node address
// if advertise address is not set // if advertise address is not set
if len(n.options.Advertise) == 0 { if len(n.options.Advertise) == 0 {
n.node.address = n.Tunnel.Address() n.node.address = n.tunnel.Address()
n.server.Init(server.Advertise(n.Tunnel.Address())) n.server.Init(server.Advertise(n.tunnel.Address()))
} }
// initialize the tunnel to resolved nodes // initialize the tunnel to resolved nodes
n.Tunnel.Init( n.tunnel.Init(
tunnel.Nodes(nodes...), tunnel.Nodes(nodes...),
) )
// dial into ControlChannel to send route adverts // dial into ControlChannel to send route adverts
ctrlClient, err := n.Tunnel.Dial(ControlChannel, tunnel.DialMulticast()) ctrlClient, err := n.tunnel.Dial(ControlChannel, tunnel.DialMulticast())
if err != nil { if err != nil {
n.Unlock() n.Unlock()
return err return err
@ -746,14 +741,14 @@ func (n *network) Connect() error {
n.tunClient[ControlChannel] = ctrlClient n.tunClient[ControlChannel] = ctrlClient
// listen on ControlChannel // listen on ControlChannel
ctrlListener, err := n.Tunnel.Listen(ControlChannel) ctrlListener, err := n.tunnel.Listen(ControlChannel)
if err != nil { if err != nil {
n.Unlock() n.Unlock()
return err return err
} }
// dial into NetworkChannel to send network messages // dial into NetworkChannel to send network messages
netClient, err := n.Tunnel.Dial(NetworkChannel, tunnel.DialMulticast()) netClient, err := n.tunnel.Dial(NetworkChannel, tunnel.DialMulticast())
if err != nil { if err != nil {
n.Unlock() n.Unlock()
return err return err
@ -762,7 +757,7 @@ func (n *network) Connect() error {
n.tunClient[NetworkChannel] = netClient n.tunClient[NetworkChannel] = netClient
// listen on NetworkChannel // listen on NetworkChannel
netListener, err := n.Tunnel.Listen(NetworkChannel) netListener, err := n.tunnel.Listen(NetworkChannel)
if err != nil { if err != nil {
n.Unlock() n.Unlock()
return err return err
@ -832,12 +827,12 @@ func (n *network) close() error {
} }
// stop the router // stop the router
if err := n.Router.Stop(); err != nil { if err := n.router.Stop(); err != nil {
return err return err
} }
// close the tunnel // close the tunnel
if err := n.Tunnel.Close(); err != nil { if err := n.tunnel.Close(); err != nil {
return err return err
} }