Merge remote-tracking branch 'origin/master'

src/broker/modules/init.go

@@ -4,35 +4,29 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/fatih/color"
+	_ "github.com/fatih/color"
 	"log"
 	"net"
 	"net/rpc"
 )
 
 // TODO: RM 처리와 init.go 전반적으로 DI 이용하여 구현 RM 키등록 부분 구현 예정
-//temporary type for matching manager
-//type match_manager struct{}
-
-func (match_mng *match_manager) matching(queue *MsgQueue) {
-	//msg := queue.pop(true)
-	//Implement here ~~
-}
-
-func (match_mng *match_manager) add_subscription(msg MsgUnit) {
-
-}
-
-//temporary type for secure(key) manager
-type secure_manager struct{}
 
 type Moscato struct {
 	queue               MsgQueue
+	SendQueue			chan myType
 	MicroServiceManager NodeManager `inject:""`
 	MatchingManager     match_manager
+	Subscriptionmanager	sub_manager
 	SecureManager       SecurityManager `inject:""`
 }
 
+type myType struct{
+	subList	[]string
+	pubMsg	MsgUnit
+	err		error
+}
+
 type Reply struct {
 	CompleteLog string
 }
@@ -144,7 +138,7 @@ func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 		moscato.queue.push(msg.(PublishMsg))
 
 	case SM: //Subscription msg
-		moscato.MatchingManager.add_subscription(msg.(*SubscriptionMsg))
+		moscato.Subscriptionmanager.addSubscription(msg.(*SubscriptionMsg))
 
 	case RM: //Register msg
 		log.Println("RM received")
@@ -217,7 +211,7 @@ func (moscato *Moscato) Run() {
 	}
 
 	//go routine -> matching 동작
-	go moscato.MatchingManager.matching(&moscato.queue)
+	go moscato.Matching()
 	go moscato.CheckQueue()
 
 	//rpc 등록 -> Receive 함수
@@ -228,7 +222,7 @@ func (moscato *Moscato) Run() {
 	}
 
 	go Listen()
-	color.Blue("initializing complete.")
+	//color.Blue("initializing complete.")
 	fmt.Scanln()
 }
 
@@ -254,4 +248,4 @@ func Listen() {
 		conn, _ := l2.Accept()
 		go rpc.ServeConn(conn)
 	}*/
-}
+}

src/broker/modules/matching.go

@@ -12,11 +12,11 @@ type match_manager struct {
 }
 
 // Matching -> Return (list of IP addresses of matched subs, Pub Msg, error)
-func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager) ([]string, MsgUnit, error) {
-	msg := queue.pop(true)
+func (moscato * Moscato) Matching() {
+	msg := moscato.queue.pop(true)
 	topic := msg.(*PublishMsg).Topic
 	value := msg.(*PublishMsg).Value
-
+	sub_mng := moscato.Subscriptionmanager
 	// list
 	ret := make([]string, 0)
 
@@ -30,127 +30,128 @@ func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager)
 
 	// Don't Exist topicNode
 	if pos == nil {
-		return nil, nil, errors.New("Don't Exist topicNode")
-	}
-
-	// 2. Traverse all valueNode -> and Match
-	valPtr := pos.list.head
-	for valPtr != nil {
-		compare := Compare(valPtr.val, value)
-		if compare < 0 { // sub.val > pub.val
-			// single : { >, >= }
-
-			// (1) case : >
-			for i := 0; i < len(valPtr.single2sub_b); i++ {
-				sub := valPtr.single2sub_b[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-
-			// (2) case : >=
-			for i := 0; i < len(valPtr.single2sub_eb); i++ {
-				sub := valPtr.single2sub_eb[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-			// range : { >, >= }
-
-			// (1) case : >
-			for i := 0; i < len(valPtr.range2sub_b); i++ {
-				sub := valPtr.range2sub_b[i]
-				big = append(big, sub)
-			}
-
-			// (2) case : >=
-			for i := 0; i < len(valPtr.range2sub_eb); i++ {
-				sub := valPtr.range2sub_eb[i]
-				big = append(big, sub)
-			}
-
-		} else if compare > 0 { // sub.val < pub.val
-
-			// single : { <, <= }
-
-			// (1) case : <
-			for i := 0; i < len(valPtr.single2sub_s); i++ {
-				sub := valPtr.single2sub_s[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
+		moscato.SendQueue <- myType{nil, msg, errors.New("Don't Exist Matching Topic")}
+	} else {
+		// 2. Traverse all valueNode -> and Match
+		valPtr := pos.list.head
+		for valPtr != nil {
+			compare := Compare(valPtr.val, value)
+			if compare < 0 { // sub.val > pub.val
+				// single : { >, >= }
+
+				// (1) case : >
+				for i := 0; i < len(valPtr.single2sub_b); i++ {
+					sub := valPtr.single2sub_b[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// (2) case : >=
+				for i := 0; i < len(valPtr.single2sub_eb); i++ {
+					sub := valPtr.single2sub_eb[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+				// range : { >, >= }
+
+				// (1) case : >
+				for i := 0; i < len(valPtr.range2sub_b); i++ {
+					sub := valPtr.range2sub_b[i]
+					big = append(big, sub)
+				}
+
+				// (2) case : >=
+				for i := 0; i < len(valPtr.range2sub_eb); i++ {
+					sub := valPtr.range2sub_eb[i]
+					big = append(big, sub)
+				}
+
+			} else if compare > 0 { // sub.val < pub.val
+
+				// single : { <, <= }
+
+				// (1) case : <
+				for i := 0; i < len(valPtr.single2sub_s); i++ {
+					sub := valPtr.single2sub_s[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// (2) case : <=
+				for i := 0; i < len(valPtr.single2sub_es); i++ {
+					sub := valPtr.single2sub_es[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// range : { <, <= }
+
+				// (1) case : <
+				for i := 0; i < len(valPtr.range2sub_s); i++ {
+					sub := valPtr.range2sub_s[i]
+					small = append(small, sub)
+				}
+				// (2) case : <=
+				for i := 0; i < len(valPtr.range2sub_es); i++ {
+					sub := valPtr.range2sub_es[i]
+					small = append(small, sub)
+				}
+
+			} else { // sub.val == pub.val
+
+				// single : { <=, >=, ==}
+
+				// (1) case : <=
+				for i := 0; i < len(valPtr.single2sub_es); i++ {
+					sub := valPtr.single2sub_es[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// (2) case : >=
+				for i := 0; i < len(valPtr.single2sub_eb); i++ {
+					sub := valPtr.single2sub_eb[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// (3) case : ==
+				for i := 0; i < len(valPtr.single2sub_e); i++ {
+					sub := valPtr.single2sub_e[i]
+					ip := sub_mng.sub2ip[sub]
+					ret = append(ret, ip)
+				}
+
+				// range : { <=, >= }
+
+				// (1) case : <=
+				for i := 0; i < len(valPtr.range2sub_es); i++ {
+					sub := valPtr.range2sub_es[i]
+					small = append(small, sub)
+				}
+
+				// (2) case : >=
+				for i := 0; i < len(valPtr.range2sub_eb); i++ {
+					sub := valPtr.range2sub_eb[i]
+					big = append(big, sub)
+				}
 
-			// (2) case : <=
-			for i := 0; i < len(valPtr.single2sub_es); i++ {
-				sub := valPtr.single2sub_es[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-
-			// range : { <, <= }
-
-			// (1) case : <
-			for i := 0; i < len(valPtr.range2sub_s); i++ {
-				sub := valPtr.range2sub_s[i]
-				small = append(small, sub)
-			}
-			// (2) case : <=
-			for i := 0; i < len(valPtr.range2sub_es); i++ {
-				sub := valPtr.range2sub_es[i]
-				small = append(small, sub)
-			}
-
-		} else { // sub.val == pub.val
-
-			// single : { <=, >=, ==}
-
-			// (1) case : <=
-			for i := 0; i < len(valPtr.single2sub_es); i++ {
-				sub := valPtr.single2sub_es[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-
-			// (2) case : >=
-			for i := 0; i < len(valPtr.single2sub_eb); i++ {
-				sub := valPtr.single2sub_eb[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-
-			// (3) case : ==
-			for i := 0; i < len(valPtr.single2sub_e); i++ {
-				sub := valPtr.single2sub_e[i]
-				ip := sub_mng.sub2ip[sub]
-				ret = append(ret, ip)
-			}
-
-			// range : { <=, >= }
-
-			// (1) case : <=
-			for i := 0; i < len(valPtr.range2sub_es); i++ {
-				sub := valPtr.range2sub_es[i]
-				small = append(small, sub)
 			}
+			valPtr = valPtr.next
+		}
 
-			// (2) case : >=
-			for i := 0; i < len(valPtr.range2sub_eb); i++ {
-				sub := valPtr.range2sub_eb[i]
-				big = append(big, sub)
-			}
+		//fmt.Println("hi")
 
+		// Add the intersection IP address of two sets (large and small) to the return list
+		hash := intersect.Hash(small, big)
+		list := reflect.ValueOf(hash)
+		for i := 0; i < list.Len(); i++ {
+			sub := list.Index(i).Interface().(int)
+			ip := sub_mng.sub2ip[sub]
+			ret = append(ret, ip)
 		}
-		valPtr = valPtr.next
-	}
 
-	// Add the intersection IP address of two sets (large and small) to the return list
-	hash := intersect.Hash(small, big)
-	list := reflect.ValueOf(hash)
-	for i := 0; i < list.Len(); i++ {
-		sub := list.Index(i).Interface().(int)
-		ip := sub_mng.sub2ip[sub]
-		ret = append(ret, ip)
+		moscato.MatchingManager.match_count++
+		moscato.SendQueue <- myType{ret, msg, nil}
 	}
-
-	match_mng.match_count++
-
-	return ret, msg, nil
-}
+}