[feat] Add method to check same ip and sub information

src/broker/modules/matching.go

@@ -2,155 +2,323 @@ package modules
 
 import (
 	"errors"
-	_ "errors"
-	"github.com/juliangruber/go-intersect"
-	"reflect"
 )
 
-type match_manager struct {
-	match_count int
+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
 }
 
-// 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
+type nodeInfo struct{
+	valNodeList []*valueNode
+	topic []int64
+}
 
-	// listß
-	ret := make([]string, 0)
+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)
+}
 
-	// list for matched range subscriptions
-	big := make([]int, 0)
-	small := make([]int, 0)
+func newSubmng() *sub_manager {
+	subMng := &sub_manager{}
+	subMng.Initialize()
+	return subMng
+}
 
-	// 1. Find (topicNode[Topic] == msg.Topic) Node
-	topicPtr := sub_mng.list
-	pos := topicPtr.getTopicNodePos(topic)
+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
 
-	// 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)
-				}
+	for i := 0; i < len(subList) && canFind == false ; i++{
+		sub := subList[i]
+		nodeinfoList := manager.sub2node[sub]
 
-				// (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 : { >, >= }
+		for j := 0; j < len(nodeinfoList); j++{
+			node := nodeinfoList[j]
 
-				// (1) case : >
-				for i := 0; i < len(valPtr.range2sub_b); i++ {
-					sub := valPtr.range2sub_b[i]
-					big = append(big, sub)
-				}
+			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]
 
-				// (2) case : >=
-				for i := 0; i < len(valPtr.range2sub_eb); i++ {
-					sub := valPtr.range2sub_eb[i]
-					big = append(big, sub)
+					leftValuePtr := node.valNodeList[0]
+					rightValuePtr := node.valNodeList[1]
+
+					nodeValList := []int64{leftValuePtr.val[0], rightValuePtr.val[0]}
+
+					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)
+						}
+					}
 				}
+				if cnt == len(node.valNodeList){
+					canFind = true
+					break
+				}
+			}
+		}
+	}
+	return canFind
+}
 
-			} else if compare > 0 { // sub.val < pub.val
+// 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
+	}
+}
 
-				// single : { <, <= }
+//	### To Insert sub#
+func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 
-				// (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)
-				}
+	topic := msg.(SubscriptionMsg).Topic
+	value := msg.(SubscriptionMsg).Value
+	operator := msg.(SubscriptionMsg).Operator
+	subnumber := 0
 
-				// (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)
-				}
+	// 0.  Check the same ip and sub information
+	if manager.isDuplicated(msg) == true {
+		return errors.New("Duplicater Subscription")
+	}
 
-				// range : { <, <= }
+	// 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
+	}
 
-				// (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)
-				}
+	nameptr := manager.list.head
+	findOk := false
 
-			} else { // sub.val == pub.val
+	// * 2. Add Subscription
 
-				// single : { <=, >=, ==}
+	// Find name in namelist, add if not found
+	for nameptr != nil {
+		if Compare(nameptr.topic, topic) == 0 {
+			findOk = true
+			break
+		}
+		nameptr = nameptr.next
+	}
 
-				// (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)
-				}
+	if !findOk {
+		manager.list.addTopicNode(topic)
+		nameptr = manager.list.tail
+	}
 
-				// (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)
-				}
+	var addValNodeList []*valueNode
 
-				// range : { <=, >= }
+	// 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) }
 
-				// (1) case : <=
-				for i := 0; i < len(valPtr.range2sub_es); i++ {
-					sub := valPtr.range2sub_es[i]
-					small = append(small, sub)
-				}
+		logical_operator := operator[2]
 
-				// (2) case : >=
-				for i := 0; i < len(valPtr.range2sub_eb); i++ {
-					sub := valPtr.range2sub_eb[i]
-					big = append(big, sub)
-				}
+		// Find ValueNode = (namelist[name].list.val == Value)
+		valptr1 := nameptr.list.getValueNodePos([]int64{value[0]})
+		valptr2 := nameptr.list.getValueNodePos([]int64{value[1]})
 
-			}
-			valPtr = valPtr.next
+		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
 		}
+		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})
 
-		// 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)
+		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)
 		}
 
-		moscato.MatchingManager.match_count++
-		moscato.SendQueue <- myType{ret, msg, nil}
+		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
+	}
+	return errors.New("Don't Exist Subscription to delete")
+}