[fin]init.go rpc & reassign MM_Receive

src/broker/modules/init.go

@@ -23,6 +23,10 @@ func (match_mng *match_manager) add_subscription(msg MsgUnit) {
 //temporary type for secure(key) manager
 type secure_manager struct{}
 
+type Microservice struct {
+
+}
+
 type Moscato struct {
 	queue      MsgQueue
 	ms_mng     MStable
@@ -34,9 +38,6 @@ type Reply struct {
 	ReceiveMsg MsgUnit
 }
 
-type Listner int
-
-
 //MS에서 보내는거-rpc call 사용
 /*func (l *Listner)MM_Receive(msg MsgUnit, reply *Reply)error{
 	reply.ReceiveMsg=msg
@@ -44,41 +45,68 @@ type Listner int
 	return nil
 }*/
 
-func (moscato *Moscato)MM_Receive(msg MsgUnit, reply *Reply)error{
-	reply.ReceiveMsg=msg
-  	go moscato.Receive(reply)
+func (moscato *Moscato)MM_Receive(msg MsgUnit)error{
+  	go moscato.Receive(msg)
 	return nil
 }
 
+
 //Recieve - MM가 msg전달 받음
-func (moscato *Moscato)Receive(reply *Reply) (MsgUnit, error) {
+func (moscato *Moscato)Receive(msg MsgUnit) (MsgUnit, error) {
 	//rpc call
-	var msg_type = reply.ReceiveMsg.CheckType()
+	var msg_type = msg.CheckType()
 	//메세지 타입에 따라 다르게 처리
 	switch msg_type {
 
 	case KSM: //Key share msg
 
 	case PM: //Publish msg
-		moscato.queue.push(reply.ReceiveMsg.(*PublishMsg))
+		moscato.queue.push(msg.(*PublishMsg))
 
 	case SM: //Subscription msg
-		moscato.match_mng.add_subscription(reply.ReceiveMsg.(*SubscriptionMsg))
+		moscato.match_mng.add_subscription(msg.(*SubscriptionMsg))
 
 	case RM: //Register msg
 		//var newmsg RegisterMsg
-		var newmsg = reply.ReceiveMsg.(*RegisterMsg)
+		var newmsg = msg.(*RegisterMsg)
 		newNode := MSnode{newmsg.From(), newmsg.From()}
 		moscato.ms_mng.add_microservice(newNode)
 
 	case WM: //Withdraw msg
-		moscato.ms_mng.remove_microservice(reply.ReceiveMsg.(*WithdrawMsg).From())
+		moscato.ms_mng.remove_microservice(msg.(*WithdrawMsg).From())
 
 	default:
 		return nil, errors.New("Message type Error: Not registered message type")
 	}
 
-	return reply.ReceiveMsg, nil
+	return msg, nil
+}
+
+func handleSum(ln net.Listener){
+	for {
+		conn, err := ln.Accept() // 클라이언트가 연결되면 TCP 연결을 리턴
+		if err != nil {
+			continue
+		}
+
+		defer func(conn net.Conn) {
+			err := conn.Close()
+			if err != nil {
+
+			}
+		}(conn)             // main 함수가 끝나기 직전에 TCP 연결을 닫음
+
+		go rpc.ServeConn(conn) // RPC를 처리하는 함수를 고루틴으로 실행
+	}
+}
+
+func (moscato *Moscato)callRPC(client *rpc.Client, msg MsgUnit){
+	reply:=new(Reply)
+	err := client.Call("MicroService.MS_Receive", msg,reply) // Calc.Sum 함수 호출
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
 }
 
 
@@ -90,8 +118,45 @@ func (moscato *Moscato) Run() {
 	go moscato.match_mng.matching(&moscato.queue)
 
 	//rpc 등록 -> Receive함수
-	rpc.Register(moscato)
 	moscato.Listen()
+
+	err := rpc.Register(new(Microservice))
+	if err != nil {
+		return
+	} // Calc 타입의 인스턴스를 생성하여 RPC 서버에 등록
+	ln, err := net.Listen("tcp", ":6000") // TCP 프로토콜에 6000번 포트로 연결을 받음 local 나중에 8150
+	ln1, err1 := net.Listen("tcp", ":6001") // TCP 프로토콜에 6000번 포트로 연결을 받음 나중에 8160
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	if err1 != nil {
+		fmt.Println(err)
+		return
+	}
+	defer func(ln net.Listener) {
+		err := ln.Close()
+		if err != nil {
+
+		}
+	}(ln)             // main 함수가 종료되기 직전에 연결 대기를 닫음
+	defer func(ln1 net.Listener) {
+		err := ln1.Close()
+		if err != nil {
+
+		}
+	}(ln1) // main 함수가 종료되기 직전에 연결 대기를 닫음
+
+	/*client1, err := rpc.Dial("tcp", "127.0.0.1:6000") // RPC 서버에 연결
+	if err != nil {
+		fmt.Println(err)
+		return
+	}*/
+
+	go handleSum(ln)
+	go handleSum(ln1)
+	fmt.Scanln() // main 함수가 종료되지 않도록 대기
 }
 
 func (moscato *Moscato) Listen() {