overflow_backup_2018_0455/overflow_discovery
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader
2018-03-23 01:11:03 +09:00
parent 3ceb82880a
commit 9589b92095
7 changed files with 49 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
discovery/discovery.go
View File

@@ -7,6 +7,7 @@ import (
"git.loafle.net/commons_go/rpc"
"git.loafle.net/commons_go/util/net/cidr"
discoveryM "git.loafle.net/overflow/overflow_commons_go/modules/discovery/model"
oopcs "git.loafle.net/overflow/overflow_probe_container/service"
)
var RPCServlet rpc.Servlet
@@ -23,20 +24,20 @@ func DiscoveryDestroy() {
discoverer = nil
}
func DiscoverZone(requesterID string, dz *discoveryM.DiscoveryZone) {
discoverer.discoverZone(requesterID, dz)
func DiscoverZone(probeService *oopcs.ProbeService, requesterID string, dz *discoveryM.DiscoveryZone) {
discoverer.discoverZone(probeService, requesterID, dz)
}
func DiscoverHost(requesterID string, zone *discoveryM.Zone, dh *discoveryM.DiscoveryHost) {
discoverer.discoverHost(requesterID, zone, dh)
func DiscoverHost(probeService *oopcs.ProbeService, requesterID string, zone *discoveryM.Zone, dh *discoveryM.DiscoveryHost) {
discoverer.discoverHost(probeService, requesterID, zone, dh)
}
func DiscoverPort(requesterID string, host *discoveryM.Host, dp *discoveryM.DiscoveryPort) {
discoverer.discoverPort(requesterID, host, dp)
func DiscoverPort(probeService *oopcs.ProbeService, requesterID string, host *discoveryM.Host, dp *discoveryM.DiscoveryPort) {
discoverer.discoverPort(probeService, requesterID, host, dp)
}
func DiscoverService(requesterID string, port *discoveryM.Port, ds *discoveryM.DiscoveryService) {
discoverer.discoverService(requesterID, port, ds)
func DiscoverService(probeService *oopcs.ProbeService, requesterID string, port *discoveryM.Port, ds *discoveryM.DiscoveryService) {
discoverer.discoverService(probeService, requesterID, port, ds)
}
func Stop() {
@@ -73,7 +74,7 @@ func (d *discovery) stop() {
logging.Logger().Infof("Discovery: discovery is stopped")
}
func (d *discovery) discoverZone(requesterID string, dz *discoveryM.DiscoveryZone) {
func (d *discovery) discoverZone(probeService *oopcs.ProbeService, requesterID string, dz *discoveryM.DiscoveryZone) {
go taskScan(d,
func(resultChan chan interface{}, errChan chan error, doneChan chan struct{}, stopChan chan struct{}) {
scanZone(dz, resultChan, errChan, doneChan, stopChan)
@@ -81,7 +82,7 @@ func (d *discovery) discoverZone(requesterID string, dz *discoveryM.DiscoveryZon
func(result interface{}) {
z := result.(*discoveryM.Zone)
logging.Logger().Debugf("zone: %v", z)
d.sendResult("DiscoveryService.DiscoveredZone", requesterID, z)
probeService.Send("DiscoveryService.DiscoveredZone", requesterID, z)
if nil != dz.DiscoveryHost {
cr, _ := cidr.NewCIDRRanger(z.Network)
dh := &discoveryM.DiscoveryHost{
@@ -89,13 +90,13 @@ func (d *discovery) discoverZone(requesterID string, dz *discoveryM.DiscoveryZon
LastScanRange: cr.Last().String(),
DiscoveryPort: dz.DiscoveryHost.DiscoveryPort,
}
d.discoverHost(requesterID, z, dh)
d.discoverHost(probeService, requesterID, z, dh)
}
},
)
}
func (d *discovery) discoverHost(requesterID string, zone *discoveryM.Zone, dh *discoveryM.DiscoveryHost) {
func (d *discovery) discoverHost(probeService *oopcs.ProbeService, requesterID string, zone *discoveryM.Zone, dh *discoveryM.DiscoveryHost) {
go taskScan(d,
func(resultChan chan interface{}, errChan chan error, doneChan chan struct{}, stopChan chan struct{}) {
scanHost(zone, dh, resultChan, errChan, doneChan, stopChan)
@@ -103,15 +104,15 @@ func (d *discovery) discoverHost(requesterID string, zone *discoveryM.Zone, dh *
func(result interface{}) {
h := result.(*discoveryM.Host)
logging.Logger().Debugf("host: %v", h)
d.sendResult("DiscoveryService.DiscoveredHost", requesterID, h)
probeService.Send("DiscoveryService.DiscoveredHost", requesterID, h)
if nil != dh.DiscoveryPort {
d.discoverPort(requesterID, h, dh.DiscoveryPort)
d.discoverPort(probeService, requesterID, h, dh.DiscoveryPort)
}
},
)
}
func (d *discovery) discoverPort(requesterID string, host *discoveryM.Host, dp *discoveryM.DiscoveryPort) {
func (d *discovery) discoverPort(probeService *oopcs.ProbeService, requesterID string, host *discoveryM.Host, dp *discoveryM.DiscoveryPort) {
go taskScan(d,
func(resultChan chan interface{}, errChan chan error, doneChan chan struct{}, stopChan chan struct{}) {
scanPort(host, dp, resultChan, errChan, doneChan, stopChan)
@@ -119,36 +120,27 @@ func (d *discovery) discoverPort(requesterID string, host *discoveryM.Host, dp *
func(result interface{}) {
p := result.(*discoveryM.Port)
logging.Logger().Debugf("port: %v", p)
d.sendResult("DiscoveryService.DiscoveredPort", requesterID, p)
probeService.Send("DiscoveryService.DiscoveredPort", requesterID, p)
if nil != dp.DiscoveryService {
d.discoverService(requesterID, p, dp.DiscoveryService)
d.discoverService(probeService, requesterID, p, dp.DiscoveryService)
}
},
)
}
func (d *discovery) discoverService(requesterID string, port *discoveryM.Port, ds *discoveryM.DiscoveryService) {
func (d *discovery) discoverService(probeService *oopcs.ProbeService, requesterID string, port *discoveryM.Port, ds *discoveryM.DiscoveryService) {
go taskScan(d,
func(resultChan chan interface{}, errChan chan error, doneChan chan struct{}, stopChan chan struct{}) {
scanService(port, ds, resultChan, errChan, doneChan, stopChan)
},
func(result interface{}) {
s := result.(*discoveryM.Service)
d.sendResult("DiscoveryService.DiscoveredService", requesterID, s)
probeService.Send("DiscoveryService.DiscoveredService", requesterID, s)
logging.Logger().Debugf("service: %s(%s)[%s:%s]", s.ServiceName, s.CryptoType, port.Host.IP, port.PortNumber)
},
)
}
func (d *discovery) sendResult(method string, args ...interface{}) {
go RPCServlet.Send(method, args...)
// go notify.Notifier.Notify(method, args...)
}
func (d *discovery) sendError() {
}
func taskScan(d *discovery, task func(resultChan chan interface{}, errChan chan error, doneChan chan struct{}, stopChan chan struct{}), onResult func(result interface{})) {
d.stopWg.Add(1)
resultChan := make(chan interface{})