Refer init.go operations for understading other funcs

main.go

@@ -1,7 +1,8 @@
 package main
 
-import "fmt"
+import "modules/Moscato_Messaging_Middleware/modules"
 
 func main() {
-	fmt.Printf("??")
+	moscato := modules.Moscato{}
+	moscato.Run()
 }

modules/init.go

@@ -1 +1,93 @@
 package modules
+
+import (
+	"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
+}
+
+// Send - rpc를 이용하여 msg전송
+func Send(ipaddr string, message MsgUnit) error {
+	return nil
+}
+
+// Recieve - rpc를 이용하여 msg전달 받음(rpc call)
+func (moscato *Moscato) Recieve(msg MsgUnit) (MsgUnit, error) {
+	msg_type := msg.CheckType()
+
+	//메세지 타입에 따라 다르게 처리
+	switch msg_type {
+
+	case KSM: //Key share msg
+
+	case PM: //Publish msg
+		moscato.queue.push(msg.(*PublishMsg))
+
+	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
+}
+
+func (moscato *Moscato) Run() {
+	//모스카토 구조체 변수 초기화
+	moscato.queue.queue_init()
+
+	//go routine -> matching 동작
+	go moscato.match_mng.matching(&moscato.queue)
+
+	//rpc 등록 -> Receive함수
+	rpc.Register(moscato)
+	moscato.Listen()
+}
+
+func (moscato *Moscato) Listen() {
+	l, err := net.Listen("tcp", fmt.Sprintf(":%v", 8160))
+
+	if err != nil {
+		log.Fatal(fmt.Sprintf("Unable to listen on given port: %s", err))
+	}
+	defer l.Close()
+
+	for {
+		conn, _ := l.Accept()
+		go rpc.ServeConn(conn)
+	}
+}

modules/manage.go

@@ -15,5 +15,41 @@ func (node *MSnode) Getipaddr() string {
 
 //모든 Microservice정보 저장
 type MStable struct {
-	IpTable map[string]MSnode
+	NodeTable map[string]MSnode
+}
+
+func (manager *MStable) getIpaddr(nodename string) (string, bool) {
+	//해당 이름의 노드이름이 존재하는지 확인
+	node, exists := manager.NodeTable[nodename]
+
+	//존재하지 않는 경우 nil리턴
+	if !exists {
+		return "", false
+	} else {
+		return node.ipAddr, true
+	}
+}
+
+func (manager *MStable) add_microservice(node MSnode) bool {
+	//삽입 전 존재여부 확인
+	_, exists := manager.NodeTable[node.Getname()]
+
+	if exists {
+		return false
+	} else {
+		manager.NodeTable[node.Getname()] = node
+		return true
+	}
+}
+
+func (manager *MStable) remove_microservice(nodename string) bool {
+	//삭제 전 존재여부 확인
+	_, exists := manager.NodeTable[nodename]
+
+	if exists {
+		delete(manager.NodeTable, nodename)
+		return true
+	} else {
+		return false
+	}
 }

modules/message.go

@@ -1,45 +1,61 @@
 package modules
 
+import "encoding/json"
+
+//import "strconv"
+//*****메세지 타입 상수화
+const (
+	KGM = 1 + iota
+	KSM
+	PM
+	SM
+	RM
+	WM
+)
+
 //*****메세지 틀*****
+
 type Message struct {
-	from    string
+	from    string //ip주소
 	version string
 	time    string
-	kine    string
-	content MsgUnit
+	kind    int //종류
 }
 
 type MsgUnit interface {
 	// ConvertToJson - send 전 json형식으로 바꾸는 함수
 	ConvertToJson() ([]byte, error)
-	// Send - rpc를 이용하여 msg전송
-	Send() (Message, error)
-	// Recieve - rpc를 이용하여 msg전달 받음(rpc call)
-	Recieve() (Message, error)
+	// CheckType - Message의 타입을 알려줌
+	CheckType() int
+	// SetType - Message 객체가 생성 되었을때 종류 정하기
+	//SetType(int)
 }
 
-//**********************
-
 //*****각 메세지 형식 및 정의**********
 
 //KeyGen 명령 메세지
 type KeyGenMsg struct {
 	Message
+	iptable []string
 }
 
 //Key공유 메세지
 type KeyShareMsg struct {
 	Message
+	key string
 }
 
 //전달할 내용을 담은 메세지
 type PublishMsg struct {
 	Message
+	subscription []int64 //선호도
+	content      []int64 // 내용
 }
 
 //구독 정보를 담은 메세지
 type SubscriptionMsg struct {
 	Message
+	subscription []int64 //선호도
 }
 
 //Microservice 등록 메세지
@@ -53,3 +69,68 @@ type WithdrawMsg struct {
 }
 
 //**************************
+
+func (msg *KeyGenMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg *KeyShareMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg *PublishMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg *SubscriptionMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg *RegisterMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg *WithdrawMsg) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+func (msg Message) CheckType() int {
+	return msg.kind
+}
+
+/*
+func (msg Message)SetType(t int){
+   msg.kind=t
+}
+
+func NewMessage() *Message{
+	//m:=&Message{
+	//from: ip
+	//version: version
+	//time : time
+	//kind : kind(종류)
+	//}
+}
+*/
+
+//*********************************
+
+// Send - rpc를 이용하여 msg전송
+//func (msg *MsgUnit) Send(ipaddr string, message MsgUnit) (error){}
+
+// Recieve - rpc를 이용하여 msg전달 받음(rpc call)
+//func Recieve(msg *MsgUnit) (MsgUnit, error){ }
+
+//*********************************

modules/queue.go

@@ -3,7 +3,7 @@ package modules
 import "errors"
 
 type MsgQueue struct {
-	queue     chan Message
+	queue     chan MsgUnit
 	QueueFunc QueueOperate
 }
 
@@ -20,29 +20,29 @@ func (mq *MsgQueue) queue_init() error {
 	if mq.queue != nil && len(mq.queue) != 0 {
 		return errors.New("Queue Hadlerer Error: Already initialized.")
 	} else if mq.queue == nil {
-		mq.queue = make(chan Message)
+		mq.queue = make(chan MsgUnit, 1000)
 		return nil
 	} else {
 		close(mq.queue)
-		mq.queue = make(chan Message)
+		mq.queue = make(chan MsgUnit)
 		return nil
 	}
 }
 
-func (mq *MsgQueue) push(msg Message) bool {
+func (mq *MsgQueue) push(msg MsgUnit) bool {
 	mq.queue <- msg
 	return true
 }
 
-func (mq *MsgQueue) pop(wait bool) (Message, error) {
+func (mq *MsgQueue) pop(wait bool) MsgUnit {
 	if wait == true {
 		if len(mq.queue) == 0 {
-			return Message{}, errors.New("Queue Handler Alert: Queue is Empty")
+			return nil
 		} else {
-			return <-mq.queue, nil
+			return <-mq.queue
 		}
 	} else {
 		//queue에 데이터가 들어올 때까지 block
-		return <-mq.queue, nil
+		return <-mq.queue
 	}
 }