package modules import ( "errors" _ "errors" "fmt" "github.com/juliangruber/go-intersect" "reflect" ) type match_manager struct { match_count int } // Matching -> Return (list of IP addresses of matched subs, Pub Msg, error) func (moscato *Moscato) Matching() { msg := moscato.queue.pop(true) topic := msg.(PublishMsg).Topic value := msg.(PublishMsg).Value sub_mng := moscato.SubscriptionManager fmt.Println("matchßing st") fmt.Println("value = ", value) // listß ret := make([]string, 0) // list for matched range subscriptions big := make([]int, 0) small := make([]int, 0) // 1. Find (topicNode[Topic] == msg.Topic) Node topicPtr := sub_mng.list pos := topicPtr.getTopicNodePos(topic) // 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) } // (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) } } valPtr = valPtr.next } //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) } fmt.Println("Big =", big) fmt.Println("small =", small) fmt.Println("matching: ", ret) moscato.MatchingManager.match_count++ moscato.SendQueue <- myType{ret, msg, nil} } }