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

src/broker/modules/subscription.go

@@ -12,15 +12,20 @@ type sub_manager struct {
 	emptylist  []int                // To administrate sub #
 	ip2sub     map[string][]int     // ip2sub[ip] = sub# ...
 	sub2ip     map[int]string       // sub2ip[sub#] = ip
-	sub2node   map[int][]*valueNode // sub2node[sub#] = node_addr ...
+	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][]*valueNode)
+	manager.sub2node = make(map[int][]nodeInfo)
 	manager.israngesub = make(map[int]bool)
 }
 
@@ -30,6 +35,94 @@ func newSubmng() *sub_manager {
 	return subMng
 }
 
+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
+
+	for i := 0; i < len(subList) && canFind == false ; i++{
+		sub := subList[i]
+		nodeinfoList := manager.sub2node[sub]
+
+		for j := 0; j < len(nodeinfoList); j++{
+			node := nodeinfoList[j]
+
+			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]
+
+					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
+}
+
+// op와 같은 node내의 operator리스트에서 sub가 있다면 리턴
+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
+	}
+}
+
 //	### To Insert sub#
 func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 
@@ -37,8 +130,13 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 	value := msg.(SubscriptionMsg).Value
 	operator := msg.(SubscriptionMsg).Operator
 	subnumber := 0
-	// * 1. Mapping incoming IP address to sub #
 
+	// 0. Check if same IP & same <Topic, Value> exists
+	if manager.isDuplicated(msg) == true {
+		return errors.New("Duplicater Subscription")
+	}
+
+	// 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)
@@ -65,11 +163,14 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 		nameptr = nameptr.next
 	}
 
+
 	if !findOk {
 		manager.list.addTopicNode(topic)
 		nameptr = manager.list.tail
 	}
 
+	var addValNodeList []*valueNode
+
 	// Add Value to list[name]
 	if len(operator) == 1 { // if single expression
 		valptr := nameptr.list.getValueNodePos(value)
@@ -77,15 +178,16 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 			nameptr.list.addValueNode(value)
 			valptr = nameptr.list.tail
 		}
-
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], valptr)
+		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) }
-		logical_operator := operator[1]
+
+		logical_operator := operator[2]
 
 		// Find ValueNode = (namelist[name].list.val == Value)
 		valptr1 := nameptr.list.getValueNodePos([]int64{value[0]})
@@ -99,9 +201,10 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 			nameptr.list.addValueNode([]int64{value[1]})
 			valptr2 = nameptr.list.tail
 		}
-
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], valptr1)
-		manager.sub2node[subnumber] = append(manager.sub2node[subnumber], valptr2)
+		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
@@ -129,90 +232,93 @@ func (manager *sub_manager) delete(from string) error {
 	for i := 0; i < len(cand); i++ {
 
 		sub := cand[i]
-		node := manager.sub2node[sub]
-
-		if manager.israngesub[sub] {
-			for j := 0; j < len(node); j++ {
-				pos := findSub(node[j].range2sub_s, sub)
-				if pos != -1 {
-					node[j].range2sub_s = remove(node[j].range2sub_s, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].range2sub_es, sub)
-				if pos != -1 {
-					node[j].range2sub_es = remove(node[j].range2sub_es, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].range2sub_b, sub)
-				if pos != -1 {
-					node[j].range2sub_b = remove(node[j].range2sub_b, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].range2sub_eb, sub)
-				if pos != -1 {
-					node[j].range2sub_eb = remove(node[j].range2sub_eb, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				isempty := node[j].isEmpty()
-
-				// Delete if Value Node is empty
-				if isempty && node[j] != nil {
-					prev_node := node[j].prev
-					next_node := node[j].next
-
-					prev_node.next = node[j].next
-					next_node.prev = node[j].prev
-				}
-			}
-		} else {
-			for j := 0; j < len(node); j++ {
-				pos := findSub(node[j].single2sub_s, sub)
-				if pos != -1 {
-					node[j].single2sub_s = remove(node[j].single2sub_s, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].single2sub_es, sub)
-				if pos != -1 {
-					node[j].single2sub_es = remove(node[j].single2sub_es, pos)
-					manager.emptylist = append(manager.emptylist, sub)
+		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
+					}
 				}
-
-				pos = findSub(node[j].single2sub_b, sub)
-				if pos != -1 {
-					node[j].single2sub_b = remove(node[j].single2sub_b, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].single2sub_eb, sub)
-				if pos != -1 {
-					node[j].single2sub_eb = remove(node[j].single2sub_eb, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				pos = findSub(node[j].single2sub_e, sub)
-				if pos != -1 {
-					node[j].single2sub_e = remove(node[j].single2sub_e, pos)
-					manager.emptylist = append(manager.emptylist, sub)
-				}
-
-				isempty := node[j].isEmpty()
-
-				// Delete if Value Node is empty
-				if isempty && node[j] != nil {
-					prevNode := node[j].prev
-					nextNode := node[j].next
-
-					prevNode.next = node[j].next
-					nextNode.prev = node[j].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
 	}
-	manager.ip2sub[ip] = nil // Delete sub#s mapped to Ip address
-	return nil
-}
+	return errors.New("Don't Exist Subscription to delete")
+}