[revise] 마이크로 서비스 등록을 위한 Manage.go 수정(인터페이스 생성과 생성자 구현)과 init.go 수정

src/broker/modules/init.go

@@ -9,6 +9,7 @@ import (
 	"net/rpc"
 )
 
+// TODO: RM 처리와 init.go 전반적으로 DI 이용하여 구현
 //temporary type for matching manager
 //type match_manager struct{}
 
@@ -25,10 +26,10 @@ func (match_mng *match_manager) add_subscription(msg MsgUnit) {
 type secure_manager struct{}
 
 type Moscato struct {
-	queue      MsgQueue
-	ms_mng     MStable
-	match_mng  match_manager
-	secure_mng secure_manager
+	queue               MsgQueue
+	MicroServiceManager NodeManager
+	match_mng           match_manager
+	secure_mng          secure_manager
 }
 
 type Reply struct {
@@ -50,7 +51,7 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
 
 	case KSM:
 		var msg KeyShareMsg
-		err:=json.Unmarshal(args.JsonMsg,&msg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
 		if err != nil {
 			return err
 		}
@@ -63,20 +64,21 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
 		reply.CompleteLog = "received completely"
 	case PM:
 		var msg PublishMsg
-		err:=json.Unmarshal(args.JsonMsg,&msg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
 		if err != nil {
 			return err
 		}
+
 		go func() {
 			_, err := receiver.moscato.Receive(msg)
 			if err != nil {
 
 			}
 		}()
-		reply.CompleteLog = "received completely"
+		reply.CompleteLog = "PM received completely"
 	case SM:
 		var msg SubscriptionMsg
-		err:=json.Unmarshal(args.JsonMsg,&msg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
 		if err != nil {
 			return err
 		}
@@ -89,20 +91,20 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
 		reply.CompleteLog = "received completely"
 	case RM:
 		var msg RegisterMsg
-		err:=json.Unmarshal(args.JsonMsg,&msg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
 		if err != nil {
 			return err
 		}
 		go func() {
 			_, err := receiver.moscato.Receive(msg)
 			if err != nil {
-
+				fmt.Println(err)
 			}
 		}()
-		reply.CompleteLog = "received completely"
+		reply.CompleteLog = "RM received completely"
 	case WM:
 		var msg WithdrawMsg
-		err:=json.Unmarshal(args.JsonMsg,&msg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
 		if err != nil {
 			return err
 		}
@@ -114,9 +116,9 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
 		}()
 		reply.CompleteLog = "received completely"
 	default:
-		return errors.New("Message type Error: Not registered message type")
+		return errors.New("message type Error: Not registered message type")
 	}
-	reply.CompleteLog = "received completely"
+	//reply.CompleteLog = "received completely"
 	return nil
 }
 
@@ -132,18 +134,25 @@ func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 	case PM: //Publish msg
 		moscato.queue.push(msg.(PublishMsg))
 		log.Println("popped queue: ", moscato.queue.pop(false))
+		log.Println("PM received")
 
 	case SM: //Subscription msg
 		moscato.match_mng.add_subscription(msg.(*SubscriptionMsg))
 
 	case RM: //Register msg
-		//var newmsg RegisterMsg
-		var newmsg = msg.(*RegisterMsg)
-		newNode := MSnode{newmsg.From, newmsg.From}
-		moscato.ms_mng.add_microservice(newNode)
+		var newMsg RegisterMsg
+		newMsg = msg.(RegisterMsg)
+
+		newNode := MSNode{newMsg.From, newMsg.From}
+		moscato.MicroServiceManager.AddMicroservice(newNode)
+
+		addr, _ := moscato.MicroServiceManager.GetIpaddr(newMsg.From)
+		fmt.Println(addr)
+
+		log.Println("RM received")
 
 	case WM: //Withdraw msg
-		moscato.ms_mng.remove_microservice(msg.(*WithdrawMsg).From)
+		moscato.MicroServiceManager.RemoveMicroservice(msg.(*WithdrawMsg).From)
 
 	default:
 		return nil, errors.New("Message type Error: Not registered message type")
@@ -153,38 +162,44 @@ func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 }
 
 //Ms로 보낼때 쓸 함수
-func (moscato *Moscato)Send2MS(ipaddress string,msg MsgUnit){
-	client,err:=rpc.Dial("tcp",ipaddress+":8150")
-	if err!=nil{
+func (moscato *Moscato) Send2MS(ipaddress string, msg MsgUnit) {
+	client, err := rpc.Dial("tcp", ipaddress+":8150")
+	if err != nil {
 		fmt.Println(err)
 		return
 	}
 	defer client.Close()
 
-	reply:=new(Reply)
-	jmsg,_:=msg.ConvertToJson()
-	args:=Args{
+	reply := new(Reply)
+	jmsg, _ := msg.ConvertToJson()
+	args := Args{
 		JsonMsg: jmsg,
 		Kind:    msg.CheckType(),
 	}
-	err:=client.Call("receiver.MsReceive",args,reply)
-	if err !=nil{
+	err = client.Call("receiver.MsReceive", args, reply)
+	if err != nil {
 		fmt.Println(err)
 		return
 	}
-	fmt.Println(reply.CompleteLog)//잘 받았는지 확인 해줌
+	fmt.Println(reply.CompleteLog) //잘 받았는지 확인 해줌
 }
 
 func (moscato *Moscato) Run() {
 	//모스카토 구조체 변수 초기화
 	receiver := Receiver{moscato: moscato}
-	moscato.queue.queue_init()
+	err := moscato.queue.queue_init()
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	moscato.MicroServiceManager = NewMStable()
 
 	//go routine -> matching 동작
 	go moscato.match_mng.matching(&moscato.queue)
 
 	//rpc 등록 -> Receive 함수
-	err := rpc.Register(receiver)
+	err = rpc.Register(receiver)
 	if err != nil {
 		log.Println(err)
 		return
@@ -196,7 +211,7 @@ func (moscato *Moscato) Run() {
 
 func Listen() {
 	//l, err := net.Listen("tcp", fmt.Sprintf(":%v", 8160))
-	l, err1 := net.Listen("tcp", ":8160")//MS로 부터 받는거
+	l, err1 := net.Listen("tcp", ":8160") //MS로 부터 받는거
 	//l2, err2 := net.Listen("tcp","0.0.0.0:8150")
 	if err1 != nil {
 		log.Fatal(fmt.Sprintf("Unable to listen on given port: %s", err1))

src/broker/modules/manage.go

@@ -1,26 +1,36 @@
 package modules
 
 //각 Microservice에 대한 정보 저장 노드
-type MSnode struct {
+type MSNode struct {
 	nodeName string
 	ipAddr   string
 }
 
-func (node *MSnode) Getname() string {
+func (node *MSNode) GetName() string {
 	return node.nodeName
 }
-func (node *MSnode) Getipaddr() string {
+func (node *MSNode) GetIpaddr() string {
 	return node.ipAddr
 }
 
+type NodeManager interface {
+	GetIpaddr(nodeName string) (string, bool)
+	AddMicroservice(node MSNode) bool
+	RemoveMicroservice(nodeName string) bool
+}
+
 //모든 Microservice정보 저장
 type MStable struct {
-	NodeTable map[string]MSnode
+	NodeTable map[string]MSNode
+}
+
+func NewMStable() *MStable {
+	return &MStable{NodeTable: make(map[string]MSNode)}
 }
 
-func (manager *MStable) getIpaddr(nodename string) (string, bool) {
+func (manager *MStable) GetIpaddr(nodeName string) (string, bool) {
 	//해당 이름의 노드이름이 존재하는지 확인
-	node, exists := manager.NodeTable[nodename]
+	node, exists := manager.NodeTable[nodeName]
 
 	//존재하지 않는 경우 nil리턴
 	if !exists {
@@ -30,24 +40,24 @@ func (manager *MStable) getIpaddr(nodename string) (string, bool) {
 	}
 }
 
-func (manager *MStable) add_microservice(node MSnode) bool {
+func (manager *MStable) AddMicroservice(node MSNode) bool {
 	//삽입 전 존재여부 확인
-	_, exists := manager.NodeTable[node.Getname()]
+	_, exists := manager.NodeTable[node.GetName()]
 
 	if exists {
 		return false
 	} else {
-		manager.NodeTable[node.Getname()] = node
+		manager.NodeTable[node.GetName()] = node
 		return true
 	}
 }
 
-func (manager *MStable) remove_microservice(nodename string) bool {
+func (manager *MStable) RemoveMicroservice(nodeName string) bool {
 	//삭제 전 존재여부 확인
-	_, exists := manager.NodeTable[nodename]
+	_, exists := manager.NodeTable[nodeName]
 
 	if exists {
-		delete(manager.NodeTable, nodename)
+		delete(manager.NodeTable, nodeName)
 		return true
 	} else {
 		return false

src/broker/modules/matching.go

@@ -8,14 +8,14 @@ import (
 )
 
 type match_manager struct {
-	match_count	int
+	match_count int
 }
 
 // Matching -> Return (list of IP addresses of matched subs, Pub Msg, error)
 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
+	topic := msg.(*PublishMsg).Topic
+	value := msg.(*PublishMsg).Value
 
 	// list
 	ret := make([]string, 0)
@@ -28,112 +28,111 @@ func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager)
 	topicPtr := sub_mng.list
 	pos := topicPtr.getTopicNodePos(topic)
 
-
 	// Don't Exist topicNode
-	if pos == nil{
-		return nil,nil,errors.New("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{
+	for valPtr != nil {
 		compare := Compare(valPtr.val, value)
 		if compare < 0 { // sub.val > pub.val
-		// single : { >, >= }
+			// single : { >, >= }
 
 			// (1) case : >
-			for i := 0; i < len(valPtr.single2sub_b); i++{
+			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++{
+			for i := 0; i < len(valPtr.single2sub_eb); i++ {
 				sub := valPtr.single2sub_eb[i]
 				ip := sub_mng.sub2ip[sub]
 				ret = append(ret, ip)
 			}
-		// range : { >, >= }
+			// range : { >, >= }
 
 			// (1) case : >
-			for i := 0; i < len(valPtr.range2sub_b); i++{
+			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++{
+			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 : { <, <= }
+			// single : { <, <= }
 
 			// (1) case : <
-			for i := 0; i < len(valPtr.single2sub_s); i++{
+			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++{
+			for i := 0; i < len(valPtr.single2sub_es); i++ {
 				sub := valPtr.single2sub_es[i]
 				ip := sub_mng.sub2ip[sub]
 				ret = append(ret, ip)
 			}
 
-		// range : { <, <= }
+			// range : { <, <= }
 
 			// (1) case : <
-			for i := 0; i < len(valPtr.range2sub_s); i++{
+			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++{
+			for i := 0; i < len(valPtr.range2sub_es); i++ {
 				sub := valPtr.range2sub_es[i]
 				small = append(small, sub)
 			}
 
-		} else{ // sub.val == pub.val
+		} else { // sub.val == pub.val
 
-		// single : { <=, >=, ==}
+			// single : { <=, >=, ==}
 
 			// (1) case : <=
-			for i := 0; i < len(valPtr.single2sub_es); i++{
+			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++{
+			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++{
+			for i := 0; i < len(valPtr.single2sub_e); i++ {
 				sub := valPtr.single2sub_e[i]
 				ip := sub_mng.sub2ip[sub]
 				ret = append(ret, ip)
 			}
 
-		// range : { <=, >= }
+			// range : { <=, >= }
 
 			// (1) case : <=
-			for i := 0; i < len(valPtr.range2sub_es); i++{
+			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++{
+			for i := 0; i < len(valPtr.range2sub_eb); i++ {
 				sub := valPtr.range2sub_eb[i]
 				big = append(big, sub)
 			}
@@ -145,7 +144,7 @@ func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager)
 	// 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++{
+	for i := 0; i < list.Len(); i++ {
 		sub := list.Index(i).Interface().(int)
 		ip := sub_mng.sub2ip[sub]
 		ret = append(ret, ip)
@@ -154,4 +153,4 @@ func (match_mng *match_manager) Matching(queue *MsgQueue, sub_mng *sub_manager)
 	match_mng.match_count++
 
 	return ret, msg, nil
-}
+}

src/broker/modules/message.go

@@ -118,7 +118,7 @@ func (msg Message) CheckType() int {
 func NewKeyGenMsg(table *MStable) *KeyGenMsg {
 	m := &KeyGenMsg{}
 	for _, value := range table.NodeTable {
-		m.iptable = append(m.iptable, value.Getipaddr())
+		m.iptable = append(m.iptable, value.GetIpaddr())
 	}
 	return m
 }

src/broker/modules/subscription.go

@@ -8,10 +8,10 @@ type sub_manager struct {
 	list topicList
 
 	/* Manage sub# */
-	count_sub 	 int               // Subscription #
+	count_sub  int                  // Subscription #
 	emptylist  []int                // To administrate sub #
 	ip2sub     map[string][]int     // ip2sub[ip] = sub# ...
-	sub2ip	   map[int]string		// sub2ip[sub#] = ip
+	sub2ip     map[int]string       // sub2ip[sub#] = ip
 	sub2node   map[int][]*valueNode // sub2node[sub#] = node_addr ...
 	israngesub map[int]bool         // To manage when deleted
 }
@@ -42,14 +42,14 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 
 	if len(manager.emptylist) == 0 {
 		subnumber = manager.count_sub
-		manager.ip2sub[msg.(*SubscriptionMsg).from] = append(manager.ip2sub[msg.(*SubscriptionMsg).from], manager.count_sub)
-		manager.sub2ip[subnumber] = msg.(*SubscriptionMsg).from
+		manager.ip2sub[msg.(*SubscriptionMsg).From] = append(manager.ip2sub[msg.(*SubscriptionMsg).From], manager.count_sub)
+		manager.sub2ip[subnumber] = msg.(*SubscriptionMsg).From
 		manager.count_sub++
 	} else {
 		subnumber := manager.emptylist[len(manager.emptylist)-1]
 		manager.emptylist = manager.emptylist[:len(manager.emptylist)-1]
-		manager.ip2sub[msg.(*SubscriptionMsg).from] = append(manager.ip2sub[msg.(*SubscriptionMsg).from], subnumber)
-		manager.sub2ip[subnumber] = msg.(*SubscriptionMsg).from
+		manager.ip2sub[msg.(*SubscriptionMsg).From] = append(manager.ip2sub[msg.(*SubscriptionMsg).From], subnumber)
+		manager.sub2ip[subnumber] = msg.(*SubscriptionMsg).From
 	}
 
 	nameptr := manager.list.head
@@ -83,7 +83,7 @@ func (manager *sub_manager) addSubscription(msg MsgUnit) error {
 		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]
@@ -214,6 +214,6 @@ func (manager *sub_manager) delete(from string) error {
 			}
 		}
 	}
-	manager.ip2sub[ip]=nil // Delete sub#s mapped to Ip address
+	manager.ip2sub[ip] = nil // Delete sub#s mapped to Ip address
 	return nil
-}
+}