Merge remote-tracking branch 'origin/master'

# Conflicts:
#	src/broker/modules/init.go

src/broker/modules/init.go

@@ -128,49 +128,7 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error { //
 	//reply.CompleteLog = "received completely"
 	return nil
 }
-
-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)
-		}
-		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)
-		}
-		return subMsg
-	}
-	return nil
-}
-
-func (moscato *Moscato) SendWithEncrypt() MsgUnit {
-	for {
-		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)
-				//moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode)
-				//fmt.Println("publish: ", mt.pubMsg)
-				//moscato.Send2MS(tmpNodeIpAddr, mt.pubMsg)
-				moscato.Send2MS(tmpNodeIpAddr, moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode))
-			}
-		}
-		return nil
-	}
-}
-
-//Recieve - MM가 msg전달 받음
+//Recieve - MM가 MS로부터 메세지 전달받음
 func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 	l4g.LoadConfiguration("modules/logConfig.json")
 	//rpc call
@@ -252,7 +210,7 @@ MM→MS
 */
 
 func (moscato *Moscato) Send2MS(ipaddress string, msg MsgUnit) {
-	//log.LoadConfiguration("logConfig.json")
+	l4g.LoadConfiguration("logConfig.json")
 	client, err := rpc.Dial("tcp", ipaddress+":8150")
 	if err != nil {
 		fmt.Println(err)
@@ -277,6 +235,48 @@ func (moscato *Moscato) Send2MS(ipaddress string, msg MsgUnit) {
 	l4g.LOGGER("Test").Info(reply.CompleteLog)
 }
 
+//Matching을 용이하게 하기위한 메세지 가공 과정
+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)
+		}
+		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)
+		}
+		return subMsg
+	}
+	return nil
+}
+
+//암호화 해서 보내기
+func (moscato *Moscato) SendWithEncrypt() MsgUnit {
+	for {
+		mt := <-moscato.SendQueue
+		if mt.err == nil {
+			for index := 0; index < len(mt.subList); index++ { //sublist들을 돌면서 매세지를 encrypt하여 메세지 보냄
+				tmpNode := mt.subList[index]
+				tmpNodeIpAddr, _ := moscato.MicroServiceManager.GetIpaddr(tmpNode)
+				//moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode)
+				//fmt.Println("publish: ", mt.pubMsg)
+				//moscato.Send2MS(tmpNodeIpAddr, mt.pubMsg)
+				moscato.Send2MS(tmpNodeIpAddr, moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode))
+			}
+		}
+		return nil
+	}
+}
+
 func (moscato *Moscato) Run() {
 
 	config := AppConfig{moscato}
@@ -313,14 +313,15 @@ func (moscato *Moscato) Run() {
 
 	go Listen()
 	color.Blue("initializing complete.")
+
 	fmt.Scanln()
 }
 
 func Listen() {
 	/*
-		MS→MM일때 ⇒ port : 8160으로 열기
+	MS→MM일때 ⇒ port : 8160으로 열기
 
-		(MM이 Server, MS가 Client)
+	(MM이 Server, MS가 Client)
 	*/
 
 	l, err1 := net.Listen("tcp", ":8160")

src/broker/modules/list.go

@@ -22,7 +22,7 @@ type valueList struct {
 }
 
 type valueNode struct {
-	val  []int64 // Encrypted Topic
+	val  []int64 // Encrypted value
 	next *valueNode
 	prev *valueNode
 
@@ -40,7 +40,7 @@ type valueNode struct {
 	range2sub_eb []int // (sub_val >= pub_val) sub#s
 }
 
-// To delete slice Array
+// To delete element in slice Array
 func remove(ary []int, i int) []int {
 	return append(ary[:i], ary[i+1:]...)
 }
@@ -56,7 +56,7 @@ func findSub(ary []int, sub int) int {
 }
 
 
-// Compare ->  To compare (topics, values) in a node
+// Compare -> To compare two encrypted arrays
 func Compare(a []int64, b []int64) int {
 	if len(a) < len(b) {
 		return 1
@@ -104,7 +104,6 @@ func (l *topicList) addTopicNode(topic []int64) {
 	l.size++
 }
 
-
 // To add a Value node to the Value list
 func (l *valueList) addValueNode(value []int64) {
 	newValNode := &valueNode{value, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}
@@ -164,7 +163,7 @@ func (node *valueNode) insertSub(op string, sub int, issingle bool) {
 	}
 }
 
-// op와 같은 node내의 operator리스트에서 sub가 있다면 리턴
+// To check if sub# is in the Value node's operator list
 func (node *valueNode) findOperatorList(op string, sub int, issingle bool) int {
 	ret := -1
 	if issingle == true {

src/broker/modules/manage.go

@@ -7,21 +7,24 @@ import (
 
 //각 Microservice에 대한 정보 저장 노드
 type MSNode struct {
-	nodeName string
+	nodeName string //Nodename- 현재 데모에서는 IpAddress와 같음
 	ipAddr   string
 }
 
+//Nodename 반환
 func (node *MSNode) GetName() string {
 	return node.nodeName
 }
+
+//IpAddress 반환
 func (node *MSNode) GetIpaddr() string {
 	return node.ipAddr
 }
 
 type NodeManager interface {
-	GetIpaddr(nodeName string) (string, bool)
-	AddMicroservice(node MSNode) bool
-	RemoveMicroservice(nodeName string) bool
+	GetIpaddr(nodeName string) (string, bool) //IpAddress반환
+	AddMicroservice(node MSNode) bool //MS추가
+	RemoveMicroservice(nodeName string) bool //MS삭제
 }
 
 //모든 Microservice정보 저장
@@ -29,11 +32,13 @@ type MStable struct {
 	NodeTable map[string]MSNode
 }
 
+//MStable 생성자
 func NewMStable() *MStable {
 	defer fmt.Println("node manager setting complete.")
 	return &MStable{NodeTable: make(map[string]MSNode)}
 }
 
+//IpAddress반환
 func (manager *MStable) GetIpaddr(nodeName string) (string, bool) {
 	//해당 이름의 노드이름이 존재하는지 확인
 	node, exists := manager.NodeTable[nodeName]
@@ -46,24 +51,26 @@ func (manager *MStable) GetIpaddr(nodeName string) (string, bool) {
 	}
 }
 
+//MS추가
 func (manager *MStable) AddMicroservice(node MSNode) bool {
 	//삽입 전 존재여부 확인
-	_, exists := manager.NodeTable[node.GetName()]
+	_, exists := manager.NodeTable[node.GetName()] ////해당 Node의 이름이 있는지 검색
 
 	if exists {
 		return false
-	} else {
+	} else { //존재안한다면 추가 (존재할경우 이미 있는것이기 때문에 추가할 필요 없음)
 		manager.NodeTable[node.GetName()] = node
 		return true
 	}
 }
 
+//MS삭제
 func (manager *MStable) RemoveMicroservice(nodeName string) bool {
 	//삭제 전 존재여부 확인
-	_, exists := manager.NodeTable[nodeName]
+	_, exists := manager.NodeTable[nodeName] //해당 이름을 가진 Node가 있는지 검색
 
 	if exists {
-		delete(manager.NodeTable, nodeName)
+		delete(manager.NodeTable, nodeName) //존재한다면 삭제
 		log.Println("[" + nodeName + "] : service quit")
 		return true
 	} else {

src/broker/modules/matching.go

@@ -8,24 +8,30 @@ import (
 )
 
 type match_manager struct {
-	match_count int
+	match_count int // Number of successful matches
 }
 
-// Matching -> Return (list of IP addresses of matched subs, Pub Msg, error)
+// Matching operation method
+// 1. Traverse TopicList for match (msg.topic == topicNode.topic)
+// 2. Traverse TopicNode.ValueList for match (msg.value <operator> valueNode.value)
+// 3. Add ipaddr to ipList when successful matching
+// 4. Add ipaddr to ipList when range matching
+// 5. Delete Duplicated Ipaddrs
+// @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
 
-	// listß
-	ret := make([]string, 0)
+	// iplist for return
+	ipList := make([]string, 0)
 
 	// list for matched range subscriptions
 	big := make([]int, 0)
 	small := make([]int, 0)
 
-	// 1. Find (topicNode[Topic] == msg.Topic) Node
+	// Find (topicNode[Topic] == msg.Topic) Node
 	topicPtr := sub_mng.list
 	pos := topicPtr.getTopicNodePos(topic)
 
@@ -33,7 +39,7 @@ func (moscato *Moscato) Matching(msg MsgUnit) {
 	if pos == nil {
 		moscato.SendQueue <- myType{nil, msg, errors.New("Don't Exist Matching Topic")}
 	} else {
-		// 2. Traverse all valueNode -> and Match
+		// Traverse all valueNode -> and Match
 		valPtr := pos.list.head
 		for valPtr != nil {
 			compare := Compare(valPtr.val, value)
@@ -44,14 +50,14 @@ func (moscato *Moscato) Matching(msg MsgUnit) {
 				for i := 0; i < len(valPtr.single2sub_b); i++ {
 					sub := valPtr.single2sub_b[i]
 					ip := sub_mng.sub2ip[sub]
-					ret = append(ret, ip)
+					ipList = append(ipList, 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)
+					ipList = append(ipList, ip)
 				}
 				// range : { >, >= }
 
@@ -75,14 +81,14 @@ func (moscato *Moscato) Matching(msg MsgUnit) {
 				for i := 0; i < len(valPtr.single2sub_s); i++ {
 					sub := valPtr.single2sub_s[i]
 					ip := sub_mng.sub2ip[sub]
-					ret = append(ret, ip)
+					ipList = append(ipList, 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)
+					ipList = append(ipList, ip)
 				}
 
 				// range : { <, <= }
@@ -106,21 +112,21 @@ func (moscato *Moscato) Matching(msg MsgUnit) {
 				for i := 0; i < len(valPtr.single2sub_es); i++ {
 					sub := valPtr.single2sub_es[i]
 					ip := sub_mng.sub2ip[sub]
-					ret = append(ret, ip)
+					ipList = append(ipList, 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)
+					ipList = append(ipList, 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)
+					ipList = append(ipList, ip)
 				}
 
 				// range : { <=, >= }
@@ -142,17 +148,26 @@ func (moscato *Moscato) Matching(msg MsgUnit) {
 		}
 
 		// Add the intersection IP address of two sets (large and small) to the return list
-		hash := intersect.Hash(small, big)
-		list := reflect.ValueOf(hash)
+		intersectSubnumbers := intersect.Hash(small, big)
+		list := reflect.ValueOf(intersectSubnumbers)
 		for i := 0; i < list.Len(); i++ {
 			sub := list.Index(i).Interface().(int)
 			ip := sub_mng.sub2ip[sub]
-			ret = append(ret, ip)
+			ipList = append(ipList, ip)
 		}
-		//fmt.Println("big = ", big)
-		//fmt.Println("small = ", small)
-		//fmt.Println("ret = ", ret)
 		moscato.MatchingManager.match_count++
-		moscato.SendQueue <- myType{ret, msg, nil}
+
+		// To delete Duplicated Ipaddr
+		keys := make(map[string]bool)
+		retIpList := []string{}
+		for _, val := range ipList {
+			if _, saveVal := keys[val]; !saveVal{
+				keys[val] = true
+				retIpList = append(retIpList, val)
+			}
+		}
+
+		// Return {IpaddressList, PubMsg, ErrorMsg} to SendQueue
+		moscato.SendQueue <- myType{retIpList, msg, nil}
 	}
-}
+}

src/broker/modules/queue.go

@@ -11,14 +11,12 @@ type MsgQueue struct {
 }
 
 type QueueOperate interface {
-	//queue 초기화
-	queue_init() error
-	//queue push
-	push(msg Message) bool
-	//pop message (wait -> busy waiting 여부 결정)
-	pop(wait bool) (Message, error)
+	queue_init() error	//Queue 초기화 멤버함수
+	push(msg Message) bool //Queue push 멤버함수
+	pop(wait bool) (Message, error)	//Queue pop 멤버함수 (wait -> busy waiting 여부 결정)
 }
 
+//Message Queue를 초기화 해주는 함수
 func (mq *MsgQueue) queue_init() error {
 	if mq.queue != nil && len(mq.queue) != 0 {
 		return errors.New("Queue Hadlerer Error: Already initialized.")
@@ -34,11 +32,13 @@ func (mq *MsgQueue) queue_init() error {
 	}
 }
 
+//push 함수
 func (mq *MsgQueue) push(msg MsgUnit) bool {
 	mq.queue <- msg
 	return true
 }
 
+//pop 함수
 func (mq *MsgQueue) pop(block bool) MsgUnit {
 	if block == false {
 		if len(mq.queue) == 0 {

src/broker/modules/secure.go

@@ -11,6 +11,7 @@ type Security struct {
 	KeyMap map[string]string
 }
 
+//Security 생성자
 func NewSecurity() *Security {
 	security := &Security{map[string]string{}}
 	fmt.Println("security setting complete.")
@@ -28,7 +29,7 @@ type SecurityManager interface {
 	//CompareAlpha(topic1 []int64, topic2 []int64) int
 }
 
-/**
+/*
 keyShareMsg 에서 각 노드의 private 키를 받아 keyMap 에 저장
 */
 func (sc Security) RegKey(rm RegisterMsg) {
@@ -48,7 +49,7 @@ func (sc Security) GetNodeKey(nodeName string) int64 {
 	return mKey
 }
 
-/**
+/*
 reEncrypt 해서 슬라이스 반환
 */
 func (sc Security) ReEncrypt(fromKey int64, toKey int64, target []int64) []int64 {
@@ -83,6 +84,7 @@ func (sc Security) ReEncPubMsg(fromPubMsg PublishMsg, nodeName string) PublishMs
 	return toPubMsg
 }
 
+//Key제거 함수
 func (sc *Security) RemoveSecureKey(nodeName string) bool {
 	//삭제 전 존재여부 확인
 	_, exists := sc.KeyMap[nodeName]
@@ -95,46 +97,3 @@ func (sc *Security) RemoveSecureKey(nodeName string) bool {
 		return false
 	}
 }
-
-/**
-Compare 함수들은 같으면 0 다르면 -1 (비교가 필요한 경우 오름차순 1 내림차순 -1)
-*/
-//func (sc Security) CompareTopic(topic1 []int64, topic2 []int64) int {
-//	// 길이 같은지 체크
-//	if len(topic1) != len(topic2) {
-//		return -1
-//	}
-//	for i := 0; i < len(topic2); i++ {
-//		if topic1[i] != topic2[i] {
-//			return -1
-//		}
-//	}
-//	return 0
-//}
-//
-//func (sc Security) CompareDigit(topic1 int64, topic2 int64) int {
-//	if topic1 < topic2 {
-//		return 1
-//	} else if topic1 > topic2 {
-//		return -1
-//	}
-//	return 0
-//}
-//
-//func (sc Security) CompareAlpha(topic1 []int64, topic2 []int64) int {
-//	// 길이 같은지 체크
-//	if len(topic1) != len(topic2) {
-//		return -1
-//	}
-//	for i := 0; i < len(topic2); i++ {
-//		if topic1[i] != topic2[i] {
-//			return -1
-//		}
-//	}
-//	return 0
-//}
-
-// private key 생성 메세지
-//func (sc Security) keyGenPrivate() KeyGenMsg{
-//
-//}