[chore] update

src/broker/modules/matching.go

@@ -2,323 +2,155 @@ package modules
 
 import (
 	"errors"
+	_ "errors"
+	"github.com/juliangruber/go-intersect"
+	"reflect"
 )
 
-type sub_manager struct {
-	list topicList
-
-	/* Manage sub# */
-	count_sub  int                  // Subscription #
-	emptylist  []int                // To administrate sub #
-	ip2sub     map[string][]int     // ip2sub[ip] = sub# ...
-	sub2ip     map[int]string       // sub2ip[sub#] = ip
-	sub2node   map[int][]nodeInfo // sub2node[sub#] = node_addr ...
-	israngesub map[int]bool         // To manage when deleted
-}
-
-type nodeInfo struct{
-	valNodeList []*valueNode
-	topic []int64
-}
-
-func (manager *sub_manager) Initialize() {
-	// Some initialize
-	manager.ip2sub = make(map[string][]int)
-	manager.sub2ip = make(map[int]string)
-	manager.sub2node = make(map[int][]nodeInfo)
-	manager.israngesub = make(map[int]bool)
-}
-
-func newSubmng() *sub_manager {
-	subMng := &sub_manager{}
-	subMng.Initialize()
-	return subMng
+type match_manager struct {
+	match_count int
 }
 
-func (manager *sub_manager) isDuplicated(msg MsgUnit) bool{
-	from := msg.(SubscriptionMsg).From
-	topic := msg.(SubscriptionMsg).Topic
-	value := msg.(SubscriptionMsg).Value
-	operator := msg.(SubscriptionMsg).Operator
-	subList := manager.ip2sub[from]
-	canFind := false
+// Matching -> Return (list of IP addresses of matched subs, Pub Msg, error)
+func (moscato *Moscato) Matching(msg MsgUnit) {
+	//msg := moscato.queue.pop(true)
+	topic := msg.(PublishMsg).Topic
+	value := msg.(PublishMsg).Value
+	sub_mng := moscato.SubscriptionManager
 
-	for i := 0; i < len(subList) && canFind == false ; i++{
-		sub := subList[i]
-		nodeinfoList := manager.sub2node[sub]
+	// listß
+	ret := make([]string, 0)
 
-		for j := 0; j < len(nodeinfoList); j++{
-			node := nodeinfoList[j]
+	// list for matched range subscriptions
+	big := make([]int, 0)
+	small := make([]int, 0)
 
-			if Compare(node.topic, topic) == 0 {
-				cnt := 0
-				if len(operator) == 1{
-					valPtr := node.valNodeList[0]
-					if Compare(valPtr.val ,value) == 0{
-						op := operator[0]
-						cnt += findOperatorList(op, valPtr, sub, true)
-					}
-				} else { // value가 숫자일 때
-					leftop := operator[0]
-					logicalop := operator[1]
-					rightop := operator[2]
+	// 1. Find (topicNode[Topic] == msg.Topic) Node
+	topicPtr := sub_mng.list
+	pos := topicPtr.getTopicNodePos(topic)
 
-					leftValuePtr := node.valNodeList[0]
-					rightValuePtr := node.valNodeList[1]
+	// Don't Exist topicNode
+	if pos == nil {
+		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)
+				}
 
-					nodeValList := []int64{leftValuePtr.val[0], rightValuePtr.val[0]}
+				// (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 : { >, >= }
 
-					if Compare(nodeValList, value) == 0 {
-						if logicalop == "&&" {
-							cnt += findOperatorList(leftop, leftValuePtr, sub, false)
-							cnt += findOperatorList(rightop, rightValuePtr, sub, false)
-						} else {
-							cnt += findOperatorList(leftop, leftValuePtr, sub, true)
-							cnt += findOperatorList(rightop, rightValuePtr, sub, true)
-						}
-					}
+				// (1) case : >
+				for i := 0; i < len(valPtr.range2sub_b); i++ {
+					sub := valPtr.range2sub_b[i]
+					big = append(big, sub)
 				}
-				if cnt == len(node.valNodeList){
-					canFind = true
-					break
+
+				// (2) case : >=
+				for i := 0; i < len(valPtr.range2sub_eb); i++ {
+					sub := valPtr.range2sub_eb[i]
+					big = append(big, sub)
 				}
-			}
-		}
-	}
-	return canFind
-}
 
-// Returns if sub exists in the operator list in the same node as op.
-func findOperatorList(op string, node *valueNode, sub int, issingle bool) int {
-	ret := -1
-	if issingle == true {
-		switch op {
-		case "<":
-			ret = findSub(node.single2sub_s, sub)
-		case "<=":
-			ret = findSub(node.single2sub_es, sub)
-		case ">":
-			ret = findSub(node.single2sub_b, sub)
-		case ">=":
-			ret = findSub(node.single2sub_eb, sub)
-		case "==":
-			ret = findSub(node.single2sub_e, sub)
-		}
-	} else {
-		switch op {
-		case "<":
-			ret = findSub(node.range2sub_s, sub)
-		case "<=":
-			ret = findSub(node.range2sub_es, sub)
-		case ">":
-			ret = findSub(node.range2sub_b, sub)
-		case ">=":
-			ret = findSub(node.range2sub_eb, sub)
-		}
-	}
-	if ret < 0{
-		return 0
-	} else{
-		return 1
-	}
-}
+			} else if compare > 0 { // sub.val < pub.val
 
-//	### To Insert sub#
-func (manager *sub_manager) addSubscription(msg MsgUnit) error {
+				// single : { <, <= }
 
-	topic := msg.(SubscriptionMsg).Topic
-	value := msg.(SubscriptionMsg).Value
-	operator := msg.(SubscriptionMsg).Operator
-	subnumber := 0
+				// (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)
+				}
 
-	// 0.  Check the same ip and sub information
-	if manager.isDuplicated(msg) == true {
-		return errors.New("Duplicater Subscription")
-	}
+				// (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)
+				}
 
-	// 1. Mapping incoming IP address to sub #
-	if len(manager.emptylist) == 0 {
-		subnumber = manager.count_sub
-		manager.ip2sub[msg.(SubscriptionMsg).From] = append(manager.ip2sub[msg.(SubscriptionMsg).From], manager.count_sub)
-		manager.sub2ip[subnumber] = msg.(SubscriptionMsg).From
-		manager.count_sub++
-	} else {
-		subnumber := manager.emptylist[len(manager.emptylist)-1]
-		manager.emptylist = manager.emptylist[:len(manager.emptylist)-1]
-		manager.ip2sub[msg.(SubscriptionMsg).From] = append(manager.ip2sub[msg.(SubscriptionMsg).From], subnumber)
-		manager.sub2ip[subnumber] = msg.(SubscriptionMsg).From
-	}
+				// range : { <, <= }
 
-	nameptr := manager.list.head
-	findOk := false
+				// (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)
+				}
 
-	// * 2. Add Subscription
+			} else { // sub.val == pub.val
 
-	// Find name in namelist, add if not found
-	for nameptr != nil {
-		if Compare(nameptr.topic, topic) == 0 {
-			findOk = true
-			break
-		}
-		nameptr = nameptr.next
-	}
+				// 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)
+				}
 
-	if !findOk {
-		manager.list.addTopicNode(topic)
-		nameptr = manager.list.tail
-	}
+				// (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)
+				}
 
-	var addValNodeList []*valueNode
+				// (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)
+				}
 
-	// Add Value to list[name]
-	if len(operator) == 1 { // if single expression
-		valptr := nameptr.list.getValueNodePos(value)
-		if valptr == nil {
-			nameptr.list.addValueNode(value)
-			valptr = nameptr.list.tail
-		}
-		addValNodeList = append(addValNodeList, valptr)
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], nodeInfo{addValNodeList, topic})
-		valptr.insertSub(operator[0], subnumber, true)
-		return nil // AddSubscription ok
-	} else {
-		// {">" "&&" "<"}
-		// For compound expressions bounded by '&&' and '||'
-		// (ex) { (234 < x) && (x <= 1293) } , { (234 < x) || ( x < 1293) }
+				// range : { <=, >= }
 
-		logical_operator := operator[2]
+				// (1) case : <=
+				for i := 0; i < len(valPtr.range2sub_es); i++ {
+					sub := valPtr.range2sub_es[i]
+					small = append(small, sub)
+				}
 
-		// Find ValueNode = (namelist[name].list.val == Value)
-		valptr1 := nameptr.list.getValueNodePos([]int64{value[0]})
-		valptr2 := nameptr.list.getValueNodePos([]int64{value[1]})
+				// (2) case : >=
+				for i := 0; i < len(valPtr.range2sub_eb); i++ {
+					sub := valPtr.range2sub_eb[i]
+					big = append(big, sub)
+				}
 
-		if valptr1 == nil {
-			nameptr.list.addValueNode([]int64{value[0]})
-			valptr1 = nameptr.list.tail
-		}
-		if valptr2 == nil {
-			nameptr.list.addValueNode([]int64{value[1]})
-			valptr2 = nameptr.list.tail
+			}
+			valPtr = valPtr.next
 		}
-		addValNodeList = append(addValNodeList, valptr1)
-		addValNodeList = append(addValNodeList, valptr2)
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], nodeInfo{addValNodeList, topic})
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], nodeInfo{addValNodeList, topic})
 
-		if logical_operator == "&&" {
-			// If they are enclosed in '&&' -> Insert Value to range_operator_list
-			manager.israngesub[subnumber] = true
-			valptr1.insertSub(operator[0], subnumber, false)
-			valptr2.insertSub(operator[2], subnumber, false)
-
-		} else {
-			// if they are enclosed in '||' -> Insert Value to single_operator_list
-			valptr1.insertSub(operator[0], subnumber, true)
-			valptr2.insertSub(operator[2], subnumber, true)
+		// 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)
 		}
 
-		return nil // addSubscription ok
-	}
-	return errors.New("Can't addSubscription")
-}
-
-// To delete subscription
-func (manager *sub_manager) delete(from string) error {
-
-	ip := from
-	cand := manager.ip2sub[ip]
-
-	for i := 0; i < len(cand); i++ {
-
-		sub := cand[i]
-		for j := 0; j < len(manager.sub2node[sub]); j++ {
-			nodeinfo := manager.sub2node[sub][j]
-			node := nodeinfo.valNodeList
-			if manager.israngesub[sub] {
-				for k := 0; k < len(node); k++ {
-					pos := findSub(node[k].range2sub_s, sub)
-					if pos != -1 {
-						node[k].range2sub_s = remove(node[k].range2sub_s, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].range2sub_es, sub)
-					if pos != -1 {
-						node[k].range2sub_es = remove(node[k].range2sub_es, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].range2sub_b, sub)
-					if pos != -1 {
-						node[k].range2sub_b = remove(node[k].range2sub_b, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].range2sub_eb, sub)
-					if pos != -1 {
-						node[k].range2sub_eb = remove(node[k].range2sub_eb, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					isempty := node[k].isEmpty()
-
-					// Delete if Value Node is empty
-					if isempty && node[k] != nil {
-						prev_node := node[k].prev
-						next_node := node[k].next
-
-						prev_node.next = node[k].next
-						next_node.prev = node[k].prev
-					}
-				}
-			} else {
-				for k := 0; k < len(node); k++ {
-					pos := findSub(node[k].single2sub_s, sub)
-					if pos != -1 {
-						node[k].single2sub_s = remove(node[k].single2sub_s, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].single2sub_es, sub)
-					if pos != -1 {
-						node[k].single2sub_es = remove(node[k].single2sub_es, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].single2sub_b, sub)
-					if pos != -1 {
-						node[k].single2sub_b = remove(node[k].single2sub_b, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].single2sub_eb, sub)
-					if pos != -1 {
-						node[k].single2sub_eb = remove(node[k].single2sub_eb, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					pos = findSub(node[k].single2sub_e, sub)
-					if pos != -1 {
-						node[k].single2sub_e = remove(node[k].single2sub_e, pos)
-						manager.emptylist = append(manager.emptylist, sub)
-					}
-
-					isempty := node[k].isEmpty()
-
-					// Delete if Value Node is empty
-					if isempty && node[k] != nil {
-						prevNode := node[k].prev
-						nextNode := node[k].next
-
-						prevNode.next = node[k].next
-						nextNode.prev = node[k].prev
-					}
-				}
-			}
-		}
-		manager.ip2sub[ip] = nil // Delete sub#s mapped to Ip address
-		return nil
+		moscato.MatchingManager.match_count++
+		moscato.SendQueue <- myType{ret, msg, nil}
 	}
-	return errors.New("Don't Exist Subscription to delete")
 }