package modules import ( "encoding/json" "errors" "fmt" "github.com/facebookgo/inject" "log" "net" "net/rpc" ) // TODO: RM 처리와 init.go 전반적으로 DI 이용하여 구현 //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 MicroServiceManager NodeManager `inject:""` 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 = "PM 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 { fmt.Println(err) } }() reply.CompleteLog = "RM 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 } func (moscato *Moscato) CheckQueue() MsgUnit { for { pmMsg := moscato.queue.pop(true) fmt.Println("queue popped : ", pmMsg) } } //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("PM received") case SM: //Subscription msg moscato.match_mng.add_subscription(msg.(*SubscriptionMsg)) case RM: //Register msg 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.MicroServiceManager.RemoveMicroservice(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() { var graph inject.Graph err := graph.Provide( &inject.Object{Value: NewMStable()}, &inject.Object{Value: moscato}) if err != nil { fmt.Println(err) return } err = graph.Populate() if err != nil { fmt.Println(err) return } //모스카토 구조체 변수 초기화 receiver := Receiver{moscato: moscato} 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) go moscato.CheckQueue() //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) }*/ }