ing

discovery/discoverer.go

@@ -43,6 +43,7 @@ type ofDiscoverer struct {
 	stopChan chan struct{}
 	stopWg   sync.WaitGroup
 
+	requestIds   map[string]bool
 	requestQueue chan types.DiscoveryRequest
 	messageChan  chan types.DiscoveryMessage
 }
@@ -63,8 +64,8 @@ func (d *ofDiscoverer) Message() <-chan types.DiscoveryMessage {
 	return d.messageChan
 }
 
-func (d *ofDiscoverer) SendMessage(discoveryRequest types.DiscoveryRequest, messageType types.DiscoveryMessageType, data interface{}, err error) {
+func (d *ofDiscoverer) SendMessage(discoveryRequest types.DiscoveryRequest, messageType types.DiscoveryMessageType, datas ...interface{}) {
-	d.messageChan <- types.MakeDiscoveryMessage(discoveryRequest, messageType, data, err)
+	d.messageChan <- types.MakeDiscoveryMessage(discoveryRequest, messageType, datas...)
 }
 
 func (d *ofDiscoverer) Shutdown() {
@@ -84,6 +85,7 @@ func (d *ofDiscoverer) Shutdown() {
 func (d *ofDiscoverer) start() {
 	d.stopChan = make(chan struct{})
 
+	d.requestIds = make(map[string]bool, 10)
 	d.requestQueue = make(chan types.DiscoveryRequest, 10)
 	d.messageChan = make(chan types.DiscoveryMessage, 256)
 
@@ -94,17 +96,18 @@ func (d *ofDiscoverer) start() {
 func (d *ofDiscoverer) enqueue(req *ofDiscoveryRequest) {
 	select {
 	case d.requestQueue <- req:
-		d.SendMessage(req, types.DiscoveryMessageTypeQueueing, omu.Now(), nil)
+		d.requestIds[req.RequestID()] = true
+		d.SendMessage(req, types.DiscoveryMessageTypeQueueing, req.RequestID(), omu.Now())
 		go func() {
 			select {
 			case <-req.dequeue:
 			case <-time.After(20 * time.Second):
 				req.timeout = true
-				d.SendMessage(req, types.DiscoveryMessageTypeQueueingTimeout, omu.Now(), nil)
+				d.SendMessage(req, types.DiscoveryMessageTypeQueueingTimeout, omu.Now())
 			}
 		}()
 	default:
-		d.SendMessage(req, types.DiscoveryMessageTypeQueueingFailed, omu.Now(), nil)
+		d.SendMessage(req, types.DiscoveryMessageTypeQueueingFailed, omu.Now())
 		select {
 		case <-time.After(time.Microsecond * 500):
 		}
@@ -125,19 +128,27 @@ LOOP:
 			if !ok {
 				return
 			}
+
+			canceled, ok := d.requestIds[req.RequestID()]
+			delete(d.requestIds, req.RequestID())
+			if !ok || canceled {
+				req.(*ofDiscoveryRequest).release()
+				continue LOOP
+			}
+
 			if req.(*ofDiscoveryRequest).timeout {
 				req.(*ofDiscoveryRequest).release()
 				continue LOOP
 			}
 			req.(*ofDiscoveryRequest).dequeue <- true
 
-			d.SendMessage(req, types.DiscoveryMessageTypeStart, omu.Now(), nil)
+			d.SendMessage(req, types.DiscoveryMessageTypeStart, omu.Now())
 			s := session.RetainDiscoverySession()
 			d.discover(req, s)
 			select {
 			case <-time.After(time.Second * 1):
 			}
-			d.SendMessage(req, types.DiscoveryMessageTypeStop, omu.Now(), nil)
+			d.SendMessage(req, types.DiscoveryMessageTypeStop, omu.Now())
 
 			select {
 			case <-time.After(time.Millisecond * 500):
@@ -159,7 +170,7 @@ func (d *ofDiscoverer) discover(req types.DiscoveryRequest, s session.DiscoveryS
 	}
 
 	if err := s.InitWithRequest(req); nil != err {
-		d.SendMessage(req, types.DiscoveryMessageTypeError, nil, err)
+		d.SendMessage(req, types.DiscoveryMessageTypeError, err)
 		return
 	}
 
@@ -185,11 +196,11 @@ func (d *ofDiscoverer) complexDiscover(s session.DiscoverySession) {
 				}
 				switch target.(type) {
 				case *omd.Host:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeHost, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeHost, target)
 				case *omd.Port:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypePort, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypePort, target)
 				case *omd.Service:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeService, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeService, target)
 				default:
 				}
 			}
@@ -244,7 +255,7 @@ func (d *ofDiscoverer) hierarchyDiscover(s session.DiscoverySession) {
 				}
 				switch target.(type) {
 				case *omd.Host:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeHost, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeHost, target)
 					if nil != s.DiscoverPort() {
 						wg.Add(1)
 						go func() {
@@ -253,7 +264,7 @@ func (d *ofDiscoverer) hierarchyDiscover(s session.DiscoverySession) {
 						}()
 					}
 				case *omd.Port:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypePort, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypePort, target)
 					if nil != s.DiscoverService() {
 						wg.Add(1)
 						go func() {
@@ -262,7 +273,7 @@ func (d *ofDiscoverer) hierarchyDiscover(s session.DiscoverySession) {
 						}()
 					}
 				case *omd.Service:
-					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeService, target, nil)
+					d.SendMessage(s.DiscoveryRequest(), types.DiscoveryMessageTypeService, target)
 				default:
 				}
 			}

discovery/discoverer_test.go

@@ -256,8 +256,7 @@ func Test_ofDiscoverer_SendMessage(t *testing.T) {
 	type args struct {
 		discoveryRequest types.DiscoveryRequest
 		messageType      types.DiscoveryMessageType
-		data             interface{}
+		datas            []interface{}
-		err              error
 	}
 	tests := []struct {
 		name   string
@@ -274,7 +273,7 @@ func Test_ofDiscoverer_SendMessage(t *testing.T) {
 				requestQueue: tt.fields.requestQueue,
 				messageChan:  tt.fields.messageChan,
 			}
-			d.SendMessage(tt.args.discoveryRequest, tt.args.messageType, tt.args.data, tt.args.err)
+			d.SendMessage(tt.args.discoveryRequest, tt.args.messageType, tt.args.datas...)
 		})
 	}
 }

discovery/discovery-request.go

@@ -4,9 +4,11 @@ import (
 	"sync"
 
 	"git.loafle.net/overflow_scanner/probe/discovery/types"
+	uuid "github.com/satori/go.uuid"
 )
 
 type ofDiscoveryRequest struct {
+	requestID   string
 	requesterID string
 	requestType types.DiscoveryRequestType
 	params      []interface{}
@@ -15,6 +17,10 @@ type ofDiscoveryRequest struct {
 	timeout bool
 }
 
+func (dr *ofDiscoveryRequest) RequestID() string {
+	return dr.requestID
+}
+
 func (dr *ofDiscoveryRequest) RequesterID() string {
 	return dr.requesterID
 }
@@ -42,6 +48,7 @@ func retainDiscoveryRequest(requesterID string, requestType types.DiscoveryReque
 		dr = v.(*ofDiscoveryRequest)
 	}
 
+	dr.requestID = uuid.NewV4().String()
 	dr.requesterID = requesterID
 	dr.requestType = requestType
 	dr.params = params
@@ -52,6 +59,7 @@ func retainDiscoveryRequest(requesterID string, requestType types.DiscoveryReque
 }
 
 func releaseDiscoveryRequest(dr *ofDiscoveryRequest) {
+	dr.requestID = ""
 	dr.requesterID = ""
 	dr.requestType = types.DiscoveryRequestTypeNone
 	dr.params = nil

discovery/types/mock-types.go

@@ -9,11 +9,16 @@ func NewMockDiscoveryRequest(requesterID string, requestType DiscoveryRequestTyp
 }
 
 type MockDiscoveryRequest struct {
+	requestID   string
 	requesterID string
 	requestType DiscoveryRequestType
 	params      []interface{}
 }
 
+func (dr *MockDiscoveryRequest) RequestID() string {
+	return dr.requestID
+}
+
 func (dr *MockDiscoveryRequest) RequesterID() string {
 	return dr.requesterID
 }

discovery/types/types.go

@@ -24,15 +24,16 @@ const (
 	DiscoveryRequestTypeService
 )
 
-type DiscoveryMessage func() (request DiscoveryRequest, messageType DiscoveryMessageType, data interface{}, err error)
+type DiscoveryMessage func() (request DiscoveryRequest, messageType DiscoveryMessageType, datas []interface{})
 
-func MakeDiscoveryMessage(request DiscoveryRequest, messageType DiscoveryMessageType, data interface{}, err error) DiscoveryMessage {
+func MakeDiscoveryMessage(request DiscoveryRequest, messageType DiscoveryMessageType, datas ...interface{}) DiscoveryMessage {
-	return func() (DiscoveryRequest, DiscoveryMessageType, interface{}, error) {
+	return func() (DiscoveryRequest, DiscoveryMessageType, []interface{}) {
-		return request, messageType, data, err
+		return request, messageType, datas
 	}
 }
 
 type DiscoveryRequest interface {
+	RequestID() string
 	RequesterID() string
 	RequestType() DiscoveryRequestType
 	Params() []interface{}

service/DiscoveryService.go

@@ -39,7 +39,7 @@ func (s *DiscoveryService) InitService() {
 					return
 				}
 
-				request, messageType, data, err := msg()
+				request, messageType, datas := msg()
 
 				switch messageType {
 				case types.DiscoveryMessageTypeStart:
@@ -47,7 +47,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveryStart",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -57,7 +57,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveryStop",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -67,7 +67,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.Queueing",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -77,7 +77,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.QueueingFailed",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -87,7 +87,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.QueueingTimeout",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -97,7 +97,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveryError",
-							[]interface{}{&orp.Error{Code: orp.E_INTERNAL, Message: err.Error()}},
+							[]interface{}{&orp.Error{Code: orp.E_INTERNAL, Message: datas[0].(error).Error()}},
 						),
 						"/scanner",
 					)
@@ -107,7 +107,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveredHost",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -117,7 +117,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveredPort",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)
@@ -127,7 +127,7 @@ func (s *DiscoveryService) InitService() {
 						rpc.MakeRPCMessage(
 							[]string{request.RequesterID()},
 							"DiscoveryService.DiscoveredService",
-							[]interface{}{data},
+							datas,
 						),
 						"/scanner",
 					)