src/broker/modules/init.go

@@ -4,14 +4,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/fatih/color"
-	l4g "github.com/jeanphorn/log4go"
 	"net"
 	"net/rpc"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
 )
 
-// TODO: RM 처리와 init.go 전반적으로 DI 이용하여 구현 RM 키등록 부분 구현 예정
-
 type Moscato struct {
 	queue               MsgQueue
 	SendQueue           chan myType
@@ -68,7 +68,7 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error { //
 
 			}
 		}()
-		reply.CompleteLog = "received"
+		reply.CompleteLog = "KSM received"
 	case PM:
 		var msg PublishMsg
 		err := json.Unmarshal(args.JsonMsg, &msg)
@@ -131,7 +131,7 @@ func (receiver Receiver) MmReceive(args Args, reply *Reply) error { //
 
 //Recieve - MM가 MS로부터 메세지 전달받음
 func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
-	var logger = newLogger()
+	logger := NewMyLogger()
 	defer logger.Sync()
 
 	//rpc call
@@ -143,43 +143,46 @@ func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
 
 	case PM: //Publish msg
 		//log.Println("PM received")
-		logger.Info("PM received")
+		fromNodeName, _ := moscato.MicroServiceManager.GetIpaddr(msg.(PublishMsg).From)
+		logger.Info("PM received from:[" + fromNodeName + "]")
 		moscato.queue.push(moscato.preProcessMsg(msg))
 
 	case SM: //Subscription msg
 		//log.Println("SM received")
-		logger.Info("SM received")
+		fromNodeName, _ := moscato.MicroServiceManager.GetIpaddr(msg.(SubscriptionMsg).From)
+		logger.Info("SM received from:[" + fromNodeName + "]")
 		err := moscato.SubscriptionManager.addSubscription(moscato.preProcessMsg(msg))
 		if err != nil {
-			println(err)
+			logger.Warn(err.Error())
 			//return nil, err
 		}
 
 	case RM: //Register msg
-		logger.Info("RM received")
 		var newMsg RegisterMsg
 		newMsg = msg.(RegisterMsg)
+		logger.Info("RM received from:[" + newMsg.From + "]")
 
 		newNode := MSNode{newMsg.From, newMsg.From}
 		resultAddNode := moscato.MicroServiceManager.AddMicroservice(newNode)
 		if resultAddNode {
-			//log.Println("Node added successful")
-			l4g.LOGGER("Test").Info("Node added successful")
+			logger.Info("Node added successful")
 		} else {
-			l4g.LOGGER("Test").Error("Node is already added, ignore RM")
+			logger.Error("Node is already added, ignore RM")
 			//log.Println("Node is already added, ignore RM")
 			return msg, nil
 		}
 
 		addr, _ := moscato.MicroServiceManager.GetIpaddr(newMsg.From)
 		moscato.SecureManager.RegKey(newMsg)
-		fmt.Println("Registered microservice: address", addr,
-			"/ key", moscato.SecureManager.GetNodeKey(newMsg.From))
+		logger.Debug("Registered microservice: address " + addr +
+			" / key " + strconv.FormatUint(uint64(moscato.SecureManager.GetNodeKey(newMsg.From)), 10))
 
 		// ackRM 메세지 전송
 		go moscato.Send2MS(addr, newMsg)
 
 	case WM: //Withdraw msg
+		fromNodeName, _ := moscato.MicroServiceManager.GetIpaddr(msg.(RegisterMsg).From)
+		logger.Info("WM received from:[" + fromNodeName + "]")
 		//ip := msg.(WithdrawMsg).From
 		//sublist := moscato.SubscriptionManager.ip2sub[ip]
 		//fmt.Println("prev list = ", sublist)
@@ -211,6 +214,9 @@ MM→MS
 */
 
 func (moscato *Moscato) Send2MS(ipaddress string, msg MsgUnit) {
+	logger := NewMyLogger()
+	defer logger.Sync()
+
 	client, err := rpc.Dial("tcp", ipaddress+":8150")
 	if err != nil {
 		fmt.Println(err)
@@ -232,7 +238,7 @@ func (moscato *Moscato) Send2MS(ipaddress string, msg MsgUnit) {
 	//log.Println(reply.CompleteLog) //잘 받았는지 확인 해줌
 	// 마이크로 서비스에게 받은 메시지는 노란색으로 출력
 	//log.Println(reply.CompleteLog)
-	//l4g.LOGGER("Test").Info(reply.CompleteLog)
+	logger.Debug(reply.CompleteLog)
 }
 
 //Matching을 용이하게 하기위한 메세지 가공 과정
@@ -278,10 +284,27 @@ func (moscato *Moscato) SendWithEncrypt() MsgUnit {
 }
 
 func (moscato *Moscato) Run() {
+	logger := NewMyLogger()
+	defer logger.Sync()
 
 	config := AppConfig{moscato}
 	config.config()
 
+	sigs := make(chan os.Signal, 1)
+	done := make(chan bool, 1)
+
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		sig := <-sigs
+		//withDraw(message, client)
+		//fmt.Println(sig)
+		_ = sig
+		done <- true
+		logger.Info("terminate Moscato Message Middleware")
+		os.Exit(0)
+	}()
+
 	//모스카토 구조체 변수 초기화
 	receiver := Receiver{moscato: moscato}
 	err := moscato.queue.queue_init()
@@ -312,12 +335,14 @@ func (moscato *Moscato) Run() {
 	}
 
 	go Listen()
-	color.Blue("initializing complete.")
+	logger.Info("initializing complete")
 
-	fmt.Scanln()
+	<-done
 }
 
 func Listen() {
+	logger := NewMyLogger()
+	defer logger.Sync()
 	/*
 		MS→MM일때 ⇒ port : 8160으로 열기
 
@@ -327,8 +352,7 @@ func Listen() {
 	l, err1 := net.Listen("tcp", ":8160")
 
 	if err1 != nil {
-		//log.Fatal(fmt.Sprintf("Unable to listen on given port: %s", err1))
-		l4g.LOGGER("Test").Critical("Unable to listen on given port: %s", err1)
+		logger.Fatal("Unable to listen on given port: " + err1.Error())
 	}
 	defer l.Close()
 

src/broker/modules/logConfig.go

@@ -12,24 +12,24 @@ func MyEncoderConfig() zapcore.EncoderConfig {
 		LevelKey: "L",
 		NameKey:  "N",
 		//CallerKey:      "C",
-		FunctionKey:    zapcore.OmitKey,
-		MessageKey:     "M",
-		StacktraceKey:  "S",
+		FunctionKey: zapcore.OmitKey,
+		MessageKey:  "M",
+		//StacktraceKey:  "S",
 		LineEnding:     zapcore.DefaultLineEnding,
-		EncodeLevel:    zapcore.CapitalLevelEncoder,
+		EncodeLevel:    zapcore.CapitalColorLevelEncoder,
 		EncodeTime:     zapcore.ISO8601TimeEncoder,
 		EncodeDuration: zapcore.StringDurationEncoder,
 		EncodeCaller:   zapcore.ShortCallerEncoder,
 	}
 }
 
-func newLogger() *zap.Logger {
+func NewMyLogger() *zap.Logger {
 	cfg := zap.Config{
 		Level:            zap.NewAtomicLevelAt(zap.DebugLevel),
 		Development:      true,
 		Encoding:         "console",
 		EncoderConfig:    MyEncoderConfig(),
-		OutputPaths:      []string{"stdout"},
+		OutputPaths:      []string{"stdout", "./test.log"},
 		ErrorOutputPaths: []string{"stderr"},
 	}
 	logger, err := cfg.Build()

src/broker/modules/manage.go

@@ -1,10 +1,5 @@
 package modules
 
-import (
-	"fmt"
-	"log"
-)
-
 //각 Microservice에 대한 정보 저장 노드
 type MSNode struct {
 	nodeName string //Nodename- 현재 데모에서는 IpAddress와 같음
@@ -23,8 +18,8 @@ func (node *MSNode) GetIpaddr() string {
 
 type NodeManager interface {
 	GetIpaddr(nodeName string) (string, bool) //IpAddress반환
-	AddMicroservice(node MSNode) bool //MS추가
-	RemoveMicroservice(nodeName string) bool //MS삭제
+	AddMicroservice(node MSNode) bool         //MS추가
+	RemoveMicroservice(nodeName string) bool  //MS삭제
 }
 
 //모든 Microservice정보 저장
@@ -34,7 +29,10 @@ type MStable struct {
 
 //MStable 생성자
 func NewMStable() *MStable {
-	defer fmt.Println("node manager setting complete.")
+	logger := NewMyLogger()
+	defer logger.Sync()
+
+	defer logger.Debug("node manager setting complete.")
 	return &MStable{NodeTable: make(map[string]MSNode)}
 }
 
@@ -66,12 +64,14 @@ func (manager *MStable) AddMicroservice(node MSNode) bool {
 
 //MS삭제
 func (manager *MStable) RemoveMicroservice(nodeName string) bool {
+	logger := NewMyLogger()
+	defer logger.Sync()
 	//삭제 전 존재여부 확인
 	_, exists := manager.NodeTable[nodeName] //해당 이름을 가진 Node가 있는지 검색
 
 	if exists {
 		delete(manager.NodeTable, nodeName) //존재한다면 삭제
-		log.Println("[" + nodeName + "] : service quit")
+		logger.Info("[" + nodeName + "] : service quit")
 		return true
 	} else {
 		return false

src/broker/modules/queue.go

@@ -2,7 +2,6 @@ package modules
 
 import (
 	"errors"
-	"fmt"
 )
 
 type MsgQueue struct {
@@ -11,19 +10,22 @@ type MsgQueue struct {
 }
 
 type QueueOperate interface {
-	queue_init() error	//Queue 초기화 멤버함수
-	push(msg Message) bool //Queue push 멤버함수
-	pop(wait bool) (Message, error)	//Queue pop 멤버함수 (wait -> busy waiting 여부 결정)
+	queue_init() error              //Queue 초기화 멤버함수
+	push(msg Message) bool          //Queue push 멤버함수
+	pop(wait bool) (Message, error) //Queue pop 멤버함수 (wait -> busy waiting 여부 결정)
 }
 
 //Message Queue를 초기화 해주는 함수
 func (mq *MsgQueue) queue_init() error {
+	logger := NewMyLogger()
+	logger.Sync()
+
 	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 MsgUnit, 1000)
 		//log.Println("queue is initialized.")
-		fmt.Println("queue is initialized.")
+		logger.Debug("queue is initialized")
 		return nil
 	} else {
 		close(mq.queue)

src/broker/modules/secure.go

@@ -2,7 +2,6 @@ package modules
 
 import (
 	"fmt"
-	"log"
 	"strconv"
 )
 
@@ -13,8 +12,10 @@ type Security struct {
 
 //Security 생성자
 func NewSecurity() *Security {
+	logger := NewMyLogger()
+	defer logger.Sync()
 	security := &Security{map[string]string{}}
-	fmt.Println("security setting complete.")
+	defer logger.Debug("security setting complete.")
 	return security
 }
 
@@ -86,12 +87,14 @@ func (sc Security) ReEncPubMsg(fromPubMsg PublishMsg, nodeName string) PublishMs
 
 //Key제거 함수
 func (sc *Security) RemoveSecureKey(nodeName string) bool {
+	logger := NewMyLogger()
+	defer logger.Sync()
 	//삭제 전 존재여부 확인
 	_, exists := sc.KeyMap[nodeName]
 
 	if exists {
 		delete(sc.KeyMap, nodeName)
-		log.Println("[" + nodeName + "] : delete Key successful")
+		logger.Debug("[" + nodeName + "] : delete Key successful")
 		return true
 	} else {
 		return false