Moscato_Messaging_Middleware/src/broker/modules/init.go

package modules

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net"
	"net/rpc"
)

//temporary type for matching manager
//type match_manager struct{}

func (match_mng *match_manager) matching(queue *MsgQueue) {
	//msg := queue.pop(true)
	//Implement here ~~
}

func (match_mng *match_manager) add_subscription(msg MsgUnit) {

}

//temporary type for secure(key) manager
type secure_manager struct{}

type Moscato struct {
	queue      MsgQueue
	ms_mng     MStable
	match_mng  match_manager
	secure_mng secure_manager
}

type Reply struct {
	CompleteLog string
}

type Receiver struct {
	moscato *Moscato
}

type Args struct { // 매개변수
	JsonMsg []byte
	Kind    int
}

func (receiver Receiver) MmReceive(args Args, reply *Reply) error {
	// 메세지 별로 나눠서 언마샬하면 됨
	switch args.Kind {

	case KSM:
		var msg KeyShareMsg
		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"
	case PM:
		var msg PublishMsg
		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"
	case SM:
		var msg SubscriptionMsg
		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"
	case RM:
		var msg RegisterMsg
		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"
	case WM:
		var msg WithdrawMsg
		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"
	default:
		return errors.New("Message type Error: Not registered message type")
	}
	reply.CompleteLog = "received completely"
	return nil
}

//Recieve - MM가 msg전달 받음
func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
	//rpc call
	var msg_type = msg.CheckType()
	//메세지 타입에 따라 다르게 처리
	switch msg_type {

	case KSM: //Key share msg

	case PM: //Publish msg
		moscato.queue.push(msg.(PublishMsg))
		log.Println("popped queue: ", moscato.queue.pop(false))

	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)

	case WM: //Withdraw msg
		moscato.ms_mng.remove_microservice(msg.(*WithdrawMsg).From)

	default:
		return nil, errors.New("Message type Error: Not registered message type")
	}

	return msg, nil
}

//Ms로 보낼때 쓸 함수
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{
		JsonMsg: jmsg,
		Kind:    msg.CheckType(),
	}
	err:=client.Call("receiver.MsReceive",args,reply)
	if err !=nil{
		fmt.Println(err)
		return
	}
	fmt.Println(reply.CompleteLog)//잘 받았는지 확인 해줌
}

func (moscato *Moscato) Run() {
	//모스카토 구조체 변수 초기화
	receiver := Receiver{moscato: moscato}
	moscato.queue.queue_init()

	//go routine -> matching 동작
	go moscato.match_mng.matching(&moscato.queue)

	//rpc 등록 -> Receive 함수
	err := rpc.Register(receiver)
	if err != nil {
		log.Println(err)
		return
	}
	Listen()
	log.Println("listen complete.")
	fmt.Scanln()
}

func Listen() {
	//l, err := net.Listen("tcp", fmt.Sprintf(":%v", 8160))
	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))
	}
	/*defer l1.Close()

	if err2 != nil {
		log.Fatal(fmt.Sprintf("Unable to listen on given port: %s", err2))
	}
	defer l2.Close()*/

	for {
		conn, _ := l.Accept()
		go rpc.ServeConn(conn)
	}
	/*for {
		conn, _ := l2.Accept()
		go rpc.ServeConn(conn)
	}*/
}