[add] 매칭, re-encryption, 마이크로서비스 send 완료

src/broker/modules/init.go

@@ -117,11 +117,45 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
 	return nil
 }
 
-func (moscato *Moscato) CheckQueue() MsgUnit {
-	for {
-		pmMsg := moscato.queue.pop(true)
-		fmt.Println("queue popped : ", pmMsg)
+func (moscato *Moscato) preProcessMsg(originalMsg MsgUnit) MsgUnit {
+	if originalMsg.CheckType() == PM {
+		pubMsg := originalMsg.(PublishMsg)
+		for index := 0; index < len(pubMsg.Topic); index++ {
+			pubMsg.Topic[index] = pubMsg.Topic[index] - moscato.SecureManager.GetNodeKey(pubMsg.From)
+		}
+		for index := 0; index < len(pubMsg.Value); index++ {
+			pubMsg.Value[index] = pubMsg.Value[index] - moscato.SecureManager.GetNodeKey(pubMsg.From)
+		}
+		fmt.Println("preprocess : ", pubMsg)
+		return pubMsg
+	} else if originalMsg.CheckType() == SM {
+		subMsg := originalMsg.(SubscriptionMsg)
+		for index := 0; index < len(subMsg.Topic); index++ {
+			subMsg.Topic[index] = subMsg.Topic[index] - moscato.SecureManager.GetNodeKey(subMsg.From)
+		}
+		for index := 0; index < len(subMsg.Value); index++ {
+			subMsg.Value[index] = subMsg.Value[index] - moscato.SecureManager.GetNodeKey(subMsg.From)
+		}
+		fmt.Println("preprocess : ", subMsg)
+		return subMsg
 	}
+	return nil
+}
+
+func (moscato *Moscato) SendWithEncrypt() MsgUnit {
+	mt := <-moscato.SendQueue
+	fmt.Println(mt)
+	if mt.err == nil {
+		for index := 0; index < len(mt.subList); index++ {
+			tmpNode := mt.subList[index]
+			tmpNodeIpAddr, _ := moscato.MicroServiceManager.GetIpaddr(tmpNode)
+			fmt.Println("before reEnc: ", mt.pubMsg)
+			moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode)
+			fmt.Println("after reEnc: ", mt.pubMsg)
+			moscato.Send2MS(tmpNodeIpAddr, mt.pubMsg)
+		}
+	}
+	return nil
 }
 
 //Recieve - MM가 msg전달 받음
@@ -136,11 +170,11 @@ func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 	case PM: //Publish msg
 		log.Println("PM received")
 		fmt.Println("PM: ", msg)
-		moscato.queue.push(msg.(PublishMsg))
+		moscato.queue.push(moscato.preProcessMsg(msg))
 
 	case SM: //Subscription msg
 		fmt.Println("SM: ", msg)
-		moscato.SubscriptionManager.addSubscription(msg)
+		moscato.SubscriptionManager.addSubscription(moscato.preProcessMsg(msg))
 
 	case RM: //Register msg
 		log.Println("RM received")
@@ -217,6 +251,7 @@ func (moscato *Moscato) Run() {
 
 	//go routine -> matching 동작
 	go moscato.Matching()
+	go moscato.SendWithEncrypt()
 
 	//go moscato.CheckQueue()
 

src/broker/modules/matching.go

@@ -3,7 +3,6 @@ package modules
 import (
 	"errors"
 	_ "errors"
-	"fmt"
 	"github.com/juliangruber/go-intersect"
 	"reflect"
 )
@@ -19,10 +18,6 @@ func (moscato *Moscato) Matching() {
 	value := msg.(PublishMsg).Value
 	sub_mng := moscato.SubscriptionManager
 
-	fmt.Println("matchßing st")
-
-	fmt.Println("value = ", value)
-
 	// listß
 	ret := make([]string, 0)
 
@@ -146,8 +141,6 @@ func (moscato *Moscato) Matching() {
 			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)
@@ -157,10 +150,6 @@ func (moscato *Moscato) Matching() {
 			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}
 	}

src/broker/modules/subscription.go

@@ -2,7 +2,6 @@ package modules
 
 import (
 	"errors"
-	"fmt"
 )
 
 type sub_manager struct {
@@ -40,9 +39,6 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 	subnumber := 0
 	// * 1. Mapping incoming IP address to sub #
 
-	fmt.Println("add sub st")
-	fmt.Println("sub val", value)
-
 	if len(manager.emptylist) == 0 {
 		subnumber = manager.count_sub
 		manager.ip2sub[msg.(SubscriptionMsg).From] = append(manager.ip2sub[msg.(SubscriptionMsg).From], manager.count_sub)
@@ -89,18 +85,18 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 
 		// For compound expressions bounded by '&&' and '||'
 		// (ex) { (234 < x) && (x <= 1293) } , { (234 < x) || ( x < 1293) }
-		logical_operator := operator[2]
+		logical_operator := operator[1]
 
 		// Find ValueNode = (namelist[name].list.val == Value)
 		valptr1 := nameptr.list.getValueNodePos([]int64{value[0]})
-		valptr2 := nameptr.list.getValueNodePos([]int64{value[2]})
+		valptr2 := nameptr.list.getValueNodePos([]int64{value[1]})
 
 		if valptr1 == nil {
 			nameptr.list.addValueNode([]int64{value[0]})
 			valptr1 = nameptr.list.tail
 		}
 		if valptr2 == nil {
-			nameptr.list.addValueNode([]int64{value[2]})
+			nameptr.list.addValueNode([]int64{value[1]})
 			valptr2 = nameptr.list.tail
 		}
 
@@ -111,12 +107,12 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 			// If they are enclosed in '&&' -> Insert Value to range_operator_list
 			manager.israngesub[subnumber] = true
 			valptr1.insertSub(operator[0], subnumber, false)
-			valptr2.insertSub(operator[4], 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[4], subnumber, true)
+			valptr2.insertSub(operator[2], subnumber, true)
 		}
 
 		return nil // addSubscription ok