[fin] matching

src/broker/modules/matching.go

@@ -1,21 +1,153 @@
 package modules
 
 import (
+	"errors"
 	_ "errors"
-	_ "fmt"
+	"github.com/juliangruber/go-intersect"
 )
 
 type match_manager struct {
+	match_count	int
 }
 
-func (match_mng *match_manager) matching(queue *MsgQueue) error {
+// Returns a list of IP addresses of matched subs
+func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager) ([]string, MsgUnit, error) {
 	msg := queue.pop(true)
+	topic := msg.(*PublishMsg).topic
+	value := msg.(*PublishMsg).value
 
-	//Implement here ~~
-	subscription := msg.(*PublishMsg).subscription
-	content := msg.(*PublishMsg).content
+	// list
+	ret := make([]string, 0)
 
-	println(subscription, content)
+	// list for matched range subscriptions
+	big := make([]int, 0)
+	small := make([]int, 0)
 
-	return nil
+	// 1. Find (topicNode[topic] == msg.topic) Node
+	topicPtr := sub_mng.list
+	pos := topicPtr.getTopicNodePos(topic)
+
+	// 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)
+			}
+
+			// (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)
+			}
+
+		}
+	}
+
+	// Add the intersection IP address of two sets (large and small) to the return list
+	hash := intersect.Hash(small, big)
+	for i := 0; i < len(hash); i++{
+		sub := hash[i].(int)
+		ip := sub_mng.sub2ip[sub]
+		ret = append(ret, ip)
+	}
+
+	match_mng.match_count++
+
+	return ret, msg, nil
 }