init commit

.idea/.gitignore

@@ -0,0 +1,8 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml
+# Editor-based HTTP Client requests
+/httpRequests/

.idea/Moscato_Messaging_Middleware.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="WEB_MODULE" version="4">
+  <component name="Go" enabled="true" />
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/Moscato_Messaging_Middleware.iml" filepath="$PROJECT_DIR$/.idea/Moscato_Messaging_Middleware.iml" />
+    </modules>
+  </component>
+</project>

.idea/vcs.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/apex/log" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/benbjohnson/clock" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/bmizerany/assert" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/davecgh/go-spew" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/facebookgo/ensure" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/facebookgo/inject" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/facebookgo/stack" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/facebookgo/structtag" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/facebookgo/subset" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/go-kit/kit" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/go-kit/log" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/go-logfmt/logfmt" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/go-stack/stack" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/jeanphorn/log4go" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/juliangruber/go-intersect" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/kr/pretty" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/kr/text" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/mattn/go-colorable" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/mattn/go-isatty" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/pkg/errors" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/pmezard/go-difflib" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/rogpeppe/go-internal" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/rs/zerolog" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/sirupsen/logrus" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/stretchr/testify" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/github.com/toolkits/file" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/go.uber.org/atomic" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/go.uber.org/goleak" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/go.uber.org/multierr" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/go.uber.org/zap" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/golang.org/x/sys" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/google.golang.org/grpc" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/gopkg.in/inconshreveable/log15.v2" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/gopkg.in/yaml.v2" vcs="Git" />
+    <mapping directory="$PROJECT_DIR$/src/gopkg.in/yaml.v3" vcs="Git" />
+  </component>
+</project>

README.md

@@ -0,0 +1,7 @@
+# Moscato_Messaging_Middleware
+
+Project Moscato
+Team Messaging Middleware
+
+Implemetation Message Middleware by Golang
+Operate as Secure, Effectively  

send2MsTime.log

@@ -0,0 +1 @@
+28500

src/broker/main.go

@@ -0,0 +1,8 @@
+package main
+
+import "broker/modules"
+
+func main() {
+	moscato := modules.Moscato{}
+	moscato.Run()
+}

src/broker/modules/AppConfig.go

@@ -0,0 +1,30 @@
+package modules
+
+import (
+	"github.com/facebookgo/inject"
+)
+
+type AppConfig struct {
+	application *Moscato
+}
+
+func (app *AppConfig) config() *Moscato {
+	var graph inject.Graph
+
+	err := graph.Provide(
+		&inject.Object{Value: NewMStable()},
+		&inject.Object{Value: app.application},
+		&inject.Object{Value: NewSecurity()})
+
+	if err != nil {
+		println(err)
+		return nil
+	}
+
+	err = graph.Populate()
+	if err != nil {
+		println(err)
+		return nil
+	}
+	return app.application
+}

src/broker/modules/image_matching.go

@@ -0,0 +1,74 @@
+package modules
+
+import (
+	"errors"
+	"fmt"
+	"math"
+)
+
+func Cosine(a []float64, b []float64) (cosine float64, err error) {
+	count := 0
+	length_a := len(a)
+	length_b := len(b)
+	if length_a > length_b {
+		count = length_a
+	} else {
+		count = length_b
+	}
+	sumA := 0.0
+	s1 := 0.0
+	s2 := 0.0
+	for k := 0; k < count; k++ {
+		if k >= length_a {
+			s2 += math.Pow(b[k], 2)
+			continue
+		}
+		if k >= length_b {
+			s1 += math.Pow(a[k], 2)
+			continue
+		}
+		sumA += a[k] * b[k]
+		s1 += math.Pow(a[k], 2)
+		s2 += math.Pow(b[k], 2)
+	}
+	if s1 == 0 || s2 == 0 {
+		return 0.0, errors.New("Vectors should not be null (all zeros)")
+	}
+	return sumA / (math.Sqrt(s1) * math.Sqrt(s2)), nil
+}
+
+func (moscato *Moscato) ImageMatching(msg MsgUnit) {
+
+	fmt.Println("gogoman")
+	topic := msg.(PublishedImage).Topic
+	sub_mng := moscato.SubscriptionManager
+
+	// iplist for return
+	//ipList := make([]string, 0)
+
+	topicPtr := sub_mng.list
+	ptr := topicPtr.getTopicBySimilarity(topic)
+
+	// No topicNode
+	if len(ptr) == 0 {
+		fmt.Println("error")
+		moscato.SendQueue <- myType{nil, msg, errors.New("No Exist Matching Topic")}
+	} else {
+
+		//find exact topic match
+		fmt.Println("here1")
+		valPtr := ptr[0]
+		fmt.Println(valPtr.topic)
+		iplist := make([]string, 0)
+		for i := 0; i < topicPtr.size; i++ {
+			pnt := sub_mng.subnum2conv[i]
+			compare := ExactCompare(valPtr.topic, pnt)
+			if compare == 1 {
+				ipAddr := sub_mng.sub2ip[i]
+				iplist = append(iplist, ipAddr)
+			}
+		}
+		moscato.MatchingManager.match_count++
+		moscato.SendQueue <- myType{iplist, msg, nil}
+	}
+}

src/broker/modules/init.go

@@ -0,0 +1,378 @@
+package modules
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"net/rpc"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
+	"time"
+)
+
+type Moscato struct {
+	queue               MsgQueue
+	SendQueue           chan myType
+	MicroServiceManager NodeManager `inject:""`
+	MatchingManager     match_manager
+	SubscriptionManager sub_manager
+	SecureManager       SecurityManager `inject:""`
+}
+
+type myType struct {
+	subList []string
+	pubMsg  MsgUnit
+	err     error
+}
+
+type Reply struct { //RPC리턴값
+	CompleteLog string //제대로 받았는지 확인하는 로그
+}
+
+type Receiver struct { //RPC 서버에 등록하기 위한 변수
+	moscato *Moscato
+}
+
+type Args struct { // 매개변수
+	JsonMsg []byte
+	Kind    int
+}
+
+/*
+MS→MM
+
+-MM 실행되면 MM서버는 열려있음(MS은 자동으로 Client)(포트 8150)
+
+-Send2MM 호출 → rpc call로 MmReceive호출해서 MM로 전달(json형식)
+
+-MmReceive에서 msgType 검사 후 그것에 맞게 msgUnit으로 MM의 Receive로 보냄
+
+-MM의 Receive에서 해당 Message를 처리
+*/
+
+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 = "KSM received"
+	case PM:
+		var msg PublishedImage
+		//fmt.Println(args.JsonMsg)
+		err := json.Unmarshal(args.JsonMsg, &msg)
+		//fmt.Println(msg)
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			_, err := receiver.moscato.Receive(msg)
+			if err != nil {
+
+			}
+		}()
+		reply.CompleteLog = "PM received"
+	case SM:
+		var msg SubscriptionImage
+		err := json.Unmarshal(args.JsonMsg, &msg)
+		if err != nil {
+			return err
+		}
+		go func() {
+			_, err := receiver.moscato.Receive(msg)
+			if err != nil {
+
+			}
+		}()
+		reply.CompleteLog = "SM received"
+	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"
+	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 = "WM received"
+	default:
+		return errors.New("message type Error: Not registered message type")
+	}
+	//reply.CompleteLog = "received completely"
+	return nil
+}
+
+//Recieve - MM가 MS로부터 메세지 전달받음
+func (moscato *Moscato) Receive(msg MsgUnit) (MsgUnit, error) {
+	logger := NewMyLogger()
+	defer logger.Sync()
+
+	//rpc call
+	var msg_type = msg.CheckType()
+	//메세지 타입에 따라 다르게 처리
+	switch msg_type {
+
+	case KSM: //Key share msg
+
+	case PM: //Publish msg
+		//log.Println("PM received")
+		fromNodeName, _ := moscato.MicroServiceManager.GetIpaddr(msg.(PublishedImage).From)
+		logger.Info("PM received from:[" + fromNodeName + "]")
+		moscato.queue.push(moscato.preProcessMsg(msg))
+
+	case SM: //Subscription msg
+		//log.Println("SM received")
+		fromNodeName, _ := moscato.MicroServiceManager.GetIpaddr(msg.(SubscriptionImage).From)
+		logger.Info("SM received from:[" + fromNodeName + "]")
+		err := moscato.SubscriptionManager.addSubscription(moscato.preProcessMsg(msg))
+		if err != nil {
+			logger.Warn(err.Error())
+			//return nil, err
+		}
+
+	case RM: //Register msg
+		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 {
+			logger.Info("Node added successful")
+		} else {
+			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)
+		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.(WithdrawMsg).From)
+		logger.Info("WM received from:[" + fromNodeName + "]")
+		//ip := msg.(WithdrawMsg).From
+		//sublist := moscato.SubscriptionManager.ip2sub[ip]
+		//fmt.Println("prev list = ", sublist)
+		moscato.MicroServiceManager.RemoveMicroservice(msg.(WithdrawMsg).From)
+		moscato.SecureManager.RemoveSecureKey(msg.(WithdrawMsg).From)
+		//moscato.SubscriptionManager.delete(ip)
+		//sublist2 := moscato.SubscriptionManager.ip2sub[ip]
+		//fmt.Println("after list =", sublist2)
+
+	default:
+		return nil, errors.New("Message type Error: Not registered message type")
+	}
+
+	return msg, nil
+}
+
+//MS로 보낼때 쓸 함수
+
+/*
+MM→MS
+
+-MS 실행되면 MS서버는 열려있음(MM은 자동으로 Client)(포트 8160)
+
+-Send2MS 호출 → rpc call로 MsReceive호출해서 MS로 전달(json형식)
+
+-MsReceive에서 msgType 검사 후 그것에 맞게 msgUnit으로 MS의 Receive로 보냄
+
+-MS의 Receive에서 해당 Message를 처리
+*/
+
+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)
+		return
+	}
+	defer client.Close()
+
+	reply := new(Reply)
+	jmsg, _ := msg.ConvertToJson()
+	args := Args{
+		JsonMsg: jmsg,
+		Kind:    msg.CheckType(),
+	}
+	err = client.Call("Receiver.Receive", args, reply)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	//log.Println(reply.CompleteLog) //잘 받았는지 확인 해줌
+	// 마이크로 서비스에게 받은 메시지는 노란색으로 출력
+	//log.Println(reply.CompleteLog)
+	logger.Debug("HERE" + reply.CompleteLog)
+}
+
+//Matching을 용이하게 하기위한 메세지 가공 과정
+func (moscato *Moscato) preProcessMsg(originalMsg MsgUnit) MsgUnit {
+	if originalMsg.CheckType() == PM {
+		pubMsg := originalMsg.(PublishedImage)
+		//for index := 0; index < len(pubMsg.Topic); index++ {
+		//	pubMsg.Topic[index] = pubMsg.Topic[index] - moscato.SecureManager.GetNodeKey(pubMsg.From)
+		//}
+		//for index := 0; index < len(pubMsg.Value); index++ {
+		//	pubMsg.Value[index] = pubMsg.Value[index] - moscato.SecureManager.GetNodeKey(pubMsg.From)
+		//}
+		return pubMsg
+	} else if originalMsg.CheckType() == SM {
+		subMsg := originalMsg.(SubscriptionImage)
+		//for index := 0; index < len(subMsg.Topic); index++ {
+		//	subMsg.Topic[index] = subMsg.Topic[index] - moscato.SecureManager.GetNodeKey(subMsg.From)
+		//}
+		//for index := 0; index < len(subMsg.Value); index++ {
+		//	subMsg.Value[index] = subMsg.Value[index] - moscato.SecureManager.GetNodeKey(subMsg.From)
+		//}
+		return subMsg
+	}
+	return nil
+}
+
+//암호화 해서 보내기
+func (moscato *Moscato) SendWithEncrypt() MsgUnit {
+	for {
+		mt := <-moscato.SendQueue
+		if mt.err == nil {
+			for index := 0; index < len(mt.subList); index++ { //sublist들을 돌면서 매세지를 encrypt하여 메세지 보냄
+				tmpNode := mt.subList[index]
+				tmpNodeIpAddr, _ := moscato.MicroServiceManager.GetIpaddr(tmpNode)
+				//moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode)
+				//fmt.Println("publish: ", mt.pubMsg)
+				//moscato.Send2MS(tmpNodeIpAddr, mt.pubMsg)
+				//moscato.Send2MS(tmpNodeIpAddr, moscato.SecureManager.ReEncPubMsg(mt.pubMsg.(PublishMsg), tmpNode))
+				fmt.Println(mt.pubMsg)
+				moscato.Send2MS(tmpNodeIpAddr, mt.pubMsg.(PublishedImage))
+			}
+
+		}
+		return nil
+	}
+}
+
+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()
+	moscato.SendQueue = make(chan myType)
+	moscato.SubscriptionManager.Initialize()
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	send2MSFile, _ := os.Create("./send2MsTime.log")
+	defer send2MSFile.Close()
+	//go routine -> matching 동작
+	go func() {
+		for {
+			msg := moscato.queue.pop(true)
+			fmt.Println("gere")
+			fmt.Println(msg) // poped msg 확인
+			encStartTime := time.Now()
+			go moscato.ImageMatching(msg)
+			go moscato.SendWithEncrypt()
+			encElapsedTime := time.Since(encStartTime)
+			fmt.Printf("MM enc시간: %d\n\n", encElapsedTime.Nanoseconds())
+			fmt.Fprintln(send2MSFile, encElapsedTime.Nanoseconds())
+		}
+	}()
+
+	//go moscato.CheckQueue()
+
+	//rpc 등록 -> Receive 함수
+	err = rpc.Register(receiver)
+	if err != nil {
+		println(err)
+		return
+	}
+
+	go Listen()
+	logger.Info("initializing complete")
+
+	<-done
+}
+
+func Listen() {
+	logger := NewMyLogger()
+	defer logger.Sync()
+	/*
+		MS→MM일때 ⇒ port : 8160으로 열기
+
+		(MM이 Server, MS가 Client)
+	*/
+
+	l, err1 := net.Listen("tcp", ":8160")
+
+	if err1 != nil {
+		logger.Fatal("Unable to listen on given port: " + err1.Error())
+	}
+	defer l.Close()
+
+	for {
+		conn, _ := l.Accept()
+		go rpc.ServeConn(conn)
+	}
+}

src/broker/modules/list.go

@@ -0,0 +1,292 @@
+package modules
+
+/* List to manage Sub information */
+
+type topicList struct {
+	head *topicNode
+	tail *topicNode
+	size int
+}
+
+type topicNode struct {
+	topic []float64 // Encrypted topic
+	next  *topicNode
+	prev  *topicNode
+	list  valueList
+}
+
+type valueList struct {
+	head *valueNode
+	tail *valueNode
+	size int
+}
+
+type valueNode struct {
+	val  []int64 // Encrypted value
+	next *valueNode
+	prev *valueNode
+
+	// single //
+	single2sub_s  []int // (sub_val < pub_val) sub#s
+	single2sub_es []int // (sub_val <= pub_val) sub#s
+	single2sub_b  []int // (sub_val > pub_val) sub#s
+	single2sub_eb []int // (sub_val >= pub_val) sub#s
+	single2sub_e  []int // (sub_val == pub_val) sub#s
+
+	// range //
+	range2sub_s  []int // (sub_val < pub_val) sub#s
+	range2sub_es []int // (sub_val <= pub_val) sub#s
+	range2sub_b  []int // (sub_val > pub_val) sub#s
+	range2sub_eb []int // (sub_val >= pub_val) sub#s
+}
+
+// To delete element in slice Array
+func remove(ary []int, i int) []int {
+	return append(ary[:i], ary[i+1:]...)
+}
+
+// To find a specific sub# in a Value node
+func findSub(ary []int, sub int) int {
+	for i := 0; i < len(ary); i++ {
+		if ary[i] == sub {
+			return i
+		}
+	}
+	return -1
+}
+
+func ExactCompare(a []float64, b []float64) int {
+	if len(a) != len(b) {
+		return 0
+	} else {
+		for i := 0; i < len(a); i++ {
+			if a[i] != b[i] {
+				return 0
+			}
+		}
+	}
+	return 1
+}
+
+func ConvCompare(a []float64, b []float64) int {
+	if len(a) < len(b) {
+		return 1
+	} else if len(a) > len(b) {
+		return -1
+	} else {
+		for i := 0; i < len(a); i++ {
+			if a[i] < b[i] {
+				return 1
+			} else if a[i] > b[i] {
+				return -1
+			}
+		}
+		return 0
+	}
+}
+
+// Compare -> To compare two encrypted arrays
+func Compare(a []int64, b []int64) int {
+	if len(a) < len(b) {
+		return 1
+	} else if len(a) > len(b) {
+		return -1
+	} else {
+		for i := 0; i < len(a); i++ {
+			if a[i] < b[i] {
+				return 1
+			} else if a[i] > b[i] {
+				return -1
+			}
+		}
+		return 0
+	}
+}
+
+func (l *topicList) getTopicBySimilarity(topic []float64) []*topicNode {
+	var threshold = float64(0.9)
+	var ptr []*topicNode
+	topicPtr := l.head
+	for topicPtr != nil {
+		if len(topicPtr.topic) == 0 {
+			topicPtr = topicPtr.next
+			continue
+		}
+		cosine, _ := Cosine(topicPtr.topic, topic)
+		if cosine > threshold {
+			ptr = append(ptr, topicPtr)
+		}
+		topicPtr = topicPtr.next
+	}
+	return ptr
+}
+
+// To get the position of a Topic node with a specific Topic
+func (l *topicList) getTopicNodePos(topic []float64) *topicNode {
+	topicPtr := l.head
+	for topicPtr != nil {
+		if len(topicPtr.topic) == 0 {
+			topicPtr = topicPtr.next
+			continue
+		}
+		if ConvCompare(topicPtr.topic, topic) == 0 {
+			return topicPtr
+		}
+		topicPtr = topicPtr.next
+	}
+	return nil
+}
+
+func (l *topicList) addConvTopic(topic []float64) {
+	newNode := &topicNode{topic, nil, nil, valueList{}}
+	if l.head == nil {
+		l.head = newNode
+		l.tail = l.head
+	} else {
+		newNode.prev = l.tail
+		l.tail.next = newNode
+		l.tail = newNode
+	}
+	l.size++
+}
+
+// To add a Topic node to the Topic list
+func (l *topicList) addTopicNode(topic []float64) {
+	newNode := &topicNode{topic, nil, nil, valueList{}}
+	if l.head == nil {
+		l.head = newNode
+		l.tail = l.head
+	} else {
+		newNode.prev = l.tail
+		l.tail.next = newNode
+		l.tail = newNode
+	}
+	l.size++
+}
+
+// To add a Value node to the Value list
+func (l *valueList) addValueNode(value []int64) {
+	newValNode := &valueNode{value, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil}
+	if l.head == nil {
+		l.head = newValNode
+		l.tail = l.head
+	} else {
+		newValNode.prev = l.tail
+		l.tail.next = newValNode
+		l.tail = newValNode
+	}
+	l.size++
+}
+
+// To get the position of a Value node with a specific Value
+func (l *valueList) getValueNodePos(value []int64) *valueNode {
+	valPtr := l.head
+	for valPtr != nil {
+		if len(valPtr.val) == 0 {
+			valPtr = valPtr.next
+			continue
+		}
+		if Compare(value, valPtr.val) == 0 {
+			return valPtr
+		}
+		valPtr = valPtr.next
+	}
+	return nil
+}
+
+// To insert sub# into the Operator list of the Value node
+func (node *valueNode) insertSub(op string, sub int, issingle bool) {
+	if issingle == true {
+		switch op {
+		case "<":
+			node.single2sub_s = append(node.single2sub_s, sub)
+		case "<=":
+			node.single2sub_es = append(node.single2sub_es, sub)
+		case ">":
+			node.single2sub_b = append(node.single2sub_b, sub)
+		case ">=":
+			node.single2sub_eb = append(node.single2sub_eb, sub)
+		case "==":
+			node.single2sub_e = append(node.single2sub_e, sub)
+		}
+	} else {
+		switch op {
+		case "<":
+			node.range2sub_s = append(node.range2sub_s, sub)
+		case "<=":
+			node.range2sub_es = append(node.range2sub_es, sub)
+		case ">":
+			node.range2sub_b = append(node.range2sub_b, sub)
+		case ">=":
+			node.range2sub_eb = append(node.range2sub_eb, sub)
+		}
+	}
+}
+
+// To check if sub# is in the Value node's operator list
+func (node *valueNode) findOperatorList(op string, sub int, issingle bool) int {
+	ret := -1
+	if issingle == true {
+		switch op {
+		case "<":
+			ret = findSub(node.single2sub_s, sub)
+		case "<=":
+			ret = findSub(node.single2sub_es, sub)
+		case ">":
+			ret = findSub(node.single2sub_b, sub)
+		case ">=":
+			ret = findSub(node.single2sub_eb, sub)
+		case "==":
+			ret = findSub(node.single2sub_e, sub)
+		}
+	} else {
+		switch op {
+		case "<":
+			ret = findSub(node.range2sub_s, sub)
+		case "<=":
+			ret = findSub(node.range2sub_es, sub)
+		case ">":
+			ret = findSub(node.range2sub_b, sub)
+		case ">=":
+			ret = findSub(node.range2sub_eb, sub)
+		}
+	}
+	if ret < 0 {
+		return 0
+	} else {
+		return 1
+	}
+}
+
+// To check if a specific Value node is empty
+func (node *valueNode) isEmpty() bool {
+	empty := true
+	if len(node.single2sub_s) != 0 {
+		empty = false
+	}
+	if len(node.single2sub_es) != 0 {
+		empty = false
+	}
+	if len(node.single2sub_b) != 0 {
+		empty = false
+	}
+	if len(node.single2sub_eb) != 0 {
+		empty = false
+	}
+	if len(node.single2sub_e) != 0 {
+		empty = false
+	}
+	if len(node.range2sub_s) != 0 {
+		empty = false
+	}
+	if len(node.range2sub_es) != 0 {
+		empty = false
+	}
+	if len(node.range2sub_b) != 0 {
+		empty = false
+	}
+	if len(node.range2sub_eb) != 0 {
+		empty = false
+	}
+	return empty
+}

src/broker/modules/list_test.go

@@ -0,0 +1,174 @@
+package modules
+
+import (
+	_ "fmt"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func Test_addTopicNode(t *testing.T) {
+
+	//list := &topicList{nil, nil, 0}
+	//ary := []int64{123, 456, 789, 1234}
+	//
+	//for i := 0; i < 4; i++ {
+	//	tmp := []int64{ary[i]}
+	//	list.addTopicNode(tmp)
+	//	exp := []int64{ary[i]}
+	//
+	//	assert.Equal(t, list.tail.topic, exp, "add TopicNode failed")
+	//}
+}
+
+func Test_addValueNode(t *testing.T) {
+
+	list := &valueList{nil, nil, 0}
+	ary := []int64{123, 456, 789, 1234}
+
+	for i := 0; i < 4; i++ {
+		tmp := []int64{ary[i]}
+		list.addValueNode(tmp)
+		exp := []int64{ary[i]}
+
+		assert.Equal(t, list.tail.val, exp, "add ValueNode failed")
+	}
+}
+
+func Test_findSub(t *testing.T) {
+
+	ary := []int{1, 2, 3, 4, 5, 6}
+	for i := 0; i < 6; i++ {
+		assert.Equal(t, findSub(ary, i+1), i, "findSub is failed")
+	}
+
+	// Don't Exist
+	assert.Equal(t, findSub(ary, 7), -1, "findSub is failed")
+
+	// array = {1, 2, 3, 4, 5, 6} -> {1, 2, 4, 5, 6}
+	ary = remove(ary, 2)
+	assert.Equal(t, findSub(ary, 6), 4)
+}
+
+func Test_remove(t *testing.T) {
+	l := make([]int, 4)
+	var i int
+	for i = 0; i < 4; i++ {
+		l[i] = i
+	}
+
+	// array = {0, 1, 2, 3} -> {0, 1, 3}
+	l = remove(l, 2)
+	assert.Equal(t, l, []int{0, 1, 3}, "Array Delete is failed")
+
+	// array = {0, 1, 3} -> {1, 3}
+	l = remove(l, 0)
+	assert.Equal(t, l, []int{1, 3}, "Array Delete is failed")
+
+	// array = {1, 2} -> {1}
+	l = remove(l, 1)
+	assert.Equal(t, l, []int{1}, "Array Delete is failed")
+
+	// array = {1} -> {}
+	l = remove(l, 0)
+	assert.Equal(t, l, []int{}, "Array Delete is failed")
+}
+
+func Test_isempty(t *testing.T) {
+	node := &valueNode{next: nil, prev: nil}
+
+	// 1. One element in Node
+	node.single2sub_eb = append(node.single2sub_eb, 1)
+	assert.Equal(t, node.isEmpty(), false, "isEmpty is failed")
+
+	// 2. Empty
+	node.single2sub_eb = remove(node.single2sub_eb, 0)
+	assert.Equal(t, node.isEmpty(), true, "isEmpty is failed")
+
+	// 3. Many elements in Node
+	node.single2sub_b = append(node.single2sub_b, 1)
+	node.single2sub_eb = append(node.single2sub_eb, 1)
+	node.single2sub_s = append(node.single2sub_s, 1)
+	node.single2sub_es = append(node.single2sub_es, 1)
+	node.single2sub_e = append(node.single2sub_e, 1)
+	assert.Equal(t, node.isEmpty(), false, "isEmpty is failed")
+}
+
+func Test_insertSub(t *testing.T) {
+	node := &valueNode{next: nil, prev: nil}
+
+	// 1. Single2sub
+
+	// (1) <
+	node.insertSub("<", 1, true)
+	assert.Equal(t, 0, findSub(node.single2sub_s, 1), "singleInsertSub (<) is failed")
+
+	// (2) <=
+	node.insertSub("<=", 2, true)
+	assert.Equal(t, 0, findSub(node.single2sub_es, 2), "singleInsertSub (<=) is failed")
+
+	// (3) >
+	node.insertSub(">", 3, true)
+	assert.Equal(t, 0, findSub(node.single2sub_b, 3), "singleInsertSub (>) is failed")
+
+	// (4) >=
+	node.insertSub(">=", 4, true)
+	assert.Equal(t, 0, findSub(node.single2sub_eb, 4), "singleInsertSub (>=) is failed")
+
+	// (5) ==
+	node.insertSub("==", 5, true)
+	assert.Equal(t, 0, findSub(node.single2sub_e, 5), "singleInsertSub (==) is failed")
+
+	// 2. range2sub
+
+	// (1) <
+	node.insertSub("<", 1, false)
+	assert.Equal(t, 0, findSub(node.range2sub_s, 1), "rangeInsertSub (<) is failed")
+
+	// (2) <=
+	node.insertSub("<=", 2, false)
+	assert.Equal(t, 0, findSub(node.range2sub_es, 2), "rangeInsertSub (<=) is failed")
+
+	// (3) >
+	node.insertSub(">", 3, false)
+	assert.Equal(t, 0, findSub(node.range2sub_b, 3), "rangeInsertSub (>) is failed")
+
+	// (4) >=
+	node.insertSub(">=", 4, false)
+	assert.Equal(t, 0, findSub(node.range2sub_eb, 4), "rangeInsertSub (>=) is failed")
+
+}
+
+func Test_getTopicPos(t *testing.T) {
+	//l := topicList{}
+	//l.addTopicNode(nil)
+	//
+	//// 1. find Topic in empty topiclist
+	//if l.head.topic != nil {
+	//	fmt.Println("addTopicNode is failed")
+	//}
+	//
+	//// 2. find {{12},{34},{56},{78},{89},{90}} in topiclist
+	//ary := [][]int64{{12}, {34}, {56}, {78}, {89}, {90}}
+	//if l.getTopicNodePos(ary[0]) != nil {
+	//	fmt.Println("getTopicNodePos is failed")
+	//}
+	//
+	//for i := 0; i < 5; i++ {
+	//	l.addTopicNode(ary[i])
+	//	assert.Equal(t, ary[i], l.tail.topic)
+	//}
+
+}
+
+func Test_getValuePos(t *testing.T) {
+	l := valueList{}
+	l.addValueNode(nil)
+
+	ary := [][]int64{{12}, {34}, {56}, {78}, {89}, {90}}
+
+	// 1. find {{12},{34},{56},{78},{89},{90}} in topiclist
+	for i := 0; i < 5; i++ {
+		l.addValueNode(ary[i])
+		assert.Equal(t, ary[i], l.tail.val)
+	}
+}

src/broker/modules/logConfig.go

@@ -0,0 +1,40 @@
+package modules
+
+import (
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+func MyEncoderConfig() zapcore.EncoderConfig {
+	return zapcore.EncoderConfig{
+		// Keys can be anything except the empty string.
+		TimeKey:  "T",
+		LevelKey: "L",
+		NameKey:  "N",
+		//CallerKey:      "C",
+		FunctionKey: zapcore.OmitKey,
+		MessageKey:  "M",
+		//StacktraceKey:  "S",
+		LineEnding:     zapcore.DefaultLineEnding,
+		EncodeLevel:    zapcore.CapitalColorLevelEncoder,
+		EncodeTime:     zapcore.ISO8601TimeEncoder,
+		EncodeDuration: zapcore.StringDurationEncoder,
+		EncodeCaller:   zapcore.ShortCallerEncoder,
+	}
+}
+
+func NewMyLogger() *zap.Logger {
+	cfg := zap.Config{
+		Level:            zap.NewAtomicLevelAt(zap.DebugLevel),
+		Development:      true,
+		Encoding:         "console",
+		EncoderConfig:    MyEncoderConfig(),
+		OutputPaths:      []string{"stdout", "./test.log"},
+		ErrorOutputPaths: []string{"stderr"},
+	}
+	logger, err := cfg.Build()
+	if err != nil {
+		panic(err)
+	}
+	return logger
+}

src/broker/modules/manage.go

@@ -0,0 +1,87 @@
+package modules
+
+import "sync"
+
+//각 Microservice에 대한 정보 저장 노드
+type MSNode struct {
+	nodeName string //Nodename- 현재 데모에서는 IpAddress와 같음
+	ipAddr   string
+}
+
+//Nodename 반환
+func (node *MSNode) GetName() string {
+	return node.nodeName
+}
+
+//IpAddress 반환
+func (node *MSNode) GetIpaddr() string {
+	return node.ipAddr
+}
+
+type NodeManager interface {
+	GetIpaddr(nodeName string) (string, bool) //IpAddress반환
+	AddMicroservice(node MSNode) bool         //MS추가
+	RemoveMicroservice(nodeName string) bool  //MS삭제
+}
+
+//모든 Microservice정보 저장
+type MStable struct {
+	NodeTable map[string]MSNode
+	mu        sync.RWMutex
+}
+
+//MStable 생성자
+func NewMStable() *MStable {
+	logger := NewMyLogger()
+	defer logger.Sync()
+
+	defer logger.Debug("node manager setting complete.")
+	return &MStable{NodeTable: make(map[string]MSNode)}
+}
+
+//IpAddress반환
+func (manager *MStable) GetIpaddr(nodeName string) (string, bool) {
+	//해당 이름의 노드이름이 존재하는지 확인
+	node, exists := manager.NodeTable[nodeName]
+
+	//존재하지 않는 경우 nil리턴
+	if !exists {
+		return "", false
+	} else {
+		return node.ipAddr, true
+	}
+}
+
+//MS추가
+func (manager *MStable) AddMicroservice(node MSNode) bool {
+	//삽입 전 존재여부 확인
+	manager.mu.RLock()
+	_, exists := manager.NodeTable[node.GetName()] ////해당 Node의 이름이 있는지 검색
+	//manager.mu.Unlock()
+	manager.mu.RUnlock()
+
+	if exists {
+		return false
+	} else { //존재안한다면 추가 (존재할경우 이미 있는것이기 때문에 추가할 필요 없음)
+		//manager.mu.Lock()
+		manager.NodeTable[node.GetName()] = node
+		//manager.mu.Unlock()
+		return true
+	}
+}
+
+//MS삭제
+func (manager *MStable) RemoveMicroservice(nodeName string) bool {
+	logger := NewMyLogger()
+	defer logger.Sync()
+	//삭제 전 존재여부 확인
+	_, exists := manager.NodeTable[nodeName] //해당 이름을 가진 Node가 있는지 검색
+
+	if exists {
+		delete(manager.NodeTable, nodeName) //존재한다면 삭제
+		logger.Info("[" + nodeName + "] : service quit")
+		return true
+	} else {
+		return false
+	}
+}

src/broker/modules/matching.go

@@ -0,0 +1,170 @@
+package modules
+
+import (
+	_ "errors"
+)
+
+type match_manager struct {
+	match_count int // Number of successful matches
+}
+
+// Matching operation method
+// 1. Traverse TopicList for match (msg.topic == topicNode.topic)
+// 2. Traverse TopicNode.ValueList for match (msg.value <operator> valueNode.value)
+// 3. Add ipaddr to ipList when successful matching
+// 4. Add ipaddr to ipList when range matching
+// 5. Delete Duplicated Ipaddrs
+// @Return (list of IP addresses of matched subs, Pub Msg, error)
+func (moscato *Moscato) Matching(msg MsgUnit) {
+
+	//topic := msg.(PublishMsg).Topic
+	//value := msg.(PublishMsg).Value
+	//sub_mng := moscato.SubscriptionManager
+	//
+	//// iplist for return
+	//ipList := make([]string, 0)
+	//
+	//// list for matched range subscriptions
+	//big := make([]int, 0)
+	//small := make([]int, 0)
+	//
+	//// Find (topicNode[Topic] == msg.Topic) Node
+	//topicPtr := sub_mng.list
+	//pos := topicPtr.getTopicNodePos(topic)
+	//
+	//// Don't Exist topicNode
+	//if pos == nil {
+	//	moscato.SendQueue <- myType{nil, msg, errors.New("Don't Exist Matching Topic")}
+	//} else {
+	//	// Traverse all valueNode -> and Match
+	//	valPtr := pos.list.head
+	//	for valPtr != nil {
+	//		compare := Compare(valPtr.val, value)
+	//		if compare > 0 { // sub.val > pub.val
+	//			// single : { >, >= }
+	//
+	//			// (1) case : >
+	//			for i := 0; i < len(valPtr.single2sub_b); i++ {
+	//				sub := valPtr.single2sub_b[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// (2) case : >=
+	//			for i := 0; i < len(valPtr.single2sub_eb); i++ {
+	//				sub := valPtr.single2sub_eb[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//			// range : { >, >= }
+	//
+	//			// (1) case : >
+	//			for i := 0; i < len(valPtr.range2sub_b); i++ {
+	//				sub := valPtr.range2sub_b[i]
+	//				big = append(big, sub)
+	//			}
+	//
+	//			// (2) case : >=
+	//			for i := 0; i < len(valPtr.range2sub_eb); i++ {
+	//				sub := valPtr.range2sub_eb[i]
+	//				big = append(big, sub)
+	//			}
+	//
+	//		} else if compare < 0 { // sub.val < pub.val
+	//
+	//			// single : { <, <= }
+	//
+	//			// (1) case : <
+	//			for i := 0; i < len(valPtr.single2sub_s); i++ {
+	//				sub := valPtr.single2sub_s[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// (2) case : <=
+	//			for i := 0; i < len(valPtr.single2sub_es); i++ {
+	//				sub := valPtr.single2sub_es[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// range : { <, <= }
+	//
+	//			// (1) case : <
+	//			for i := 0; i < len(valPtr.range2sub_s); i++ {
+	//				sub := valPtr.range2sub_s[i]
+	//				small = append(small, sub)
+	//			}
+	//			// (2) case : <=
+	//			for i := 0; i < len(valPtr.range2sub_es); i++ {
+	//				sub := valPtr.range2sub_es[i]
+	//				small = append(small, sub)
+	//			}
+	//
+	//		} else { // sub.val == pub.val
+	//
+	//			// single : { <=, >=, ==}
+	//
+	//			// (1) case : <=
+	//			for i := 0; i < len(valPtr.single2sub_es); i++ {
+	//				sub := valPtr.single2sub_es[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// (2) case : >=
+	//			for i := 0; i < len(valPtr.single2sub_eb); i++ {
+	//				sub := valPtr.single2sub_eb[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// (3) case : ==
+	//			for i := 0; i < len(valPtr.single2sub_e); i++ {
+	//				sub := valPtr.single2sub_e[i]
+	//				ip := sub_mng.sub2ip[sub]
+	//				ipList = append(ipList, ip)
+	//			}
+	//
+	//			// range : { <=, >= }
+	//
+	//			// (1) case : <=
+	//			for i := 0; i < len(valPtr.range2sub_es); i++ {
+	//				sub := valPtr.range2sub_es[i]
+	//				small = append(small, sub)
+	//			}
+	//
+	//			// (2) case : >=
+	//			for i := 0; i < len(valPtr.range2sub_eb); i++ {
+	//				sub := valPtr.range2sub_eb[i]
+	//				big = append(big, sub)
+	//			}
+	//
+	//		}
+	//		valPtr = valPtr.next
+	//	}
+	//
+	//	// Add the intersection IP address of two sets (large and small) to the return list
+	//	intersectSubnumbers := intersect.Hash(small, big)
+	//	list := reflect.ValueOf(intersectSubnumbers)
+	//	for i := 0; i < list.Len(); i++ {
+	//		sub := list.Index(i).Interface().(int)
+	//		ip := sub_mng.sub2ip[sub]
+	//		ipList = append(ipList, ip)
+	//	}
+	//	moscato.MatchingManager.match_count++
+	//
+	//	// To delete Duplicated Ipaddr
+	//	keys := make(map[string]bool)
+	//	retIpList := []string{}
+	//	for _, val := range ipList {
+	//		if _, saveVal := keys[val]; !saveVal{
+	//			keys[val] = true
+	//			retIpList = append(retIpList, val)
+	//		}
+	//	}
+	//
+	//	// Return {IpaddressList, PubMsg, ErrorMsg} to SendQueue
+	//	moscato.SendQueue <- myType{retIpList, msg, nil}
+	//}
+}

src/broker/modules/matching_test.go

@@ -0,0 +1,170 @@
+package modules
+
+//
+//import (
+//	"fmt"
+//	"github.com/stretchr/testify/assert"
+//	"math/rand"
+//	"strconv"
+//	"testing"
+//	"time"
+//)
+//
+//
+//func makePubData(IsAlpha bool) MsgUnit{
+//	rand.Seed(time.Now().UnixNano())
+//	// Set Ipaddr
+//	msg := Message{"", "1.0", "", PM}
+//	for i := 0; i < 4; i++{
+//		itoa := strconv.Itoa(rand.Int() % 256)
+//		msg.From += itoa
+//		if i != 3{
+//			msg.From += "."
+//		}
+//	}
+//
+//	// Set Time
+//	msg.Time += strconv.Itoa(rand.Int()%24) + ":"
+//	msg.Time += strconv.Itoa(rand.Int()%60)
+//
+//	Topic := []int64{}
+//	Value := []int64{}
+//	content := []int64{}
+//
+//	// Set Topic
+//	topicLen := rand.Int() % 10 + 1
+//	for i := 0 ; i < topicLen; i++{
+//		Topic = append(Topic, rand.Int63())
+//	}
+//
+//	// Set Value
+//	if IsAlpha {
+//		valueLen := rand.Int() % 10 + 1
+//		for i := 0 ; i < valueLen; i++{
+//			Value = append(Value, rand.Int63())
+//		}
+//	} else {
+//		Topic = append(Topic, rand.Int63())
+//		Value = append(Value, rand.Int63())
+//	}
+//
+//	// Set content
+//	contentLen := rand.Int() % 10
+//	for i := 0; i < contentLen; i++{
+//		content = append(content, rand.Int63())
+//	}
+//
+//	return &PublishMsg{msg, Topic, Value, content}
+//}
+//
+//func makeSubData(IsAlpha bool, Topic []int64, Value []int64) MsgUnit{
+//	rand.Seed(time.Now().UnixNano())
+//	// Set Ipaddr
+//	msg := Message{"", "1.0", "", SM}
+//	for i := 0; i < 4; i++{
+//		itoa := strconv.Itoa(rand.Int() % 256)
+//		msg.From += itoa
+//		if i != 3{
+//			msg.From += "."
+//		}
+//	}
+//
+//	// Set Time
+//	msg.Time += strconv.Itoa(rand.Int()%24) + ":"
+//	msg.Time += strconv.Itoa(rand.Int()%60)
+//
+//	// Set Topic, Value, Operator
+//	Operator := []string{}
+//	candOp := []string{">", ">=", "<=" ,"<", "=="}
+//	logicalOp := []string{"&&", "||"}
+//
+//	if IsAlpha {
+//		Operator = append(Operator, "==")
+//	} else {
+//		randSeed := rand.Int() % 2
+//		if randSeed == 0 {
+//			lop := rand.Int() % 2
+//			Operator = append(Operator, candOp[rand.Int()%2])
+//			if lop == 0 {
+//				Operator = append(Operator, logicalOp[0])
+//				for {
+//					x := rand.Int63()
+//					if x > Value[0] {
+//						Value = append(Value, x)
+//						break
+//					}
+//				}
+//			} else {
+//				Operator = append(Operator, logicalOp[1])
+//				Value = append(Value, rand.Int63())
+//			}
+//			Operator = append(Operator, candOp[rand.Int()%2+2])
+//		} else {
+//			op := candOp[rand.Int()%5]
+//			Operator = append(Operator, op)
+//		}
+//	}
+//
+//	return &SubscriptionMsg{msg, Topic, Value, Operator, IsAlpha}
+//}
+//
+//func Test_matching(t *testing.T) {
+//	rand.Seed(time.Now().UnixNano())
+//	assert.Equal(t, 1, 1)
+//	mos := Moscato{ sub_mng: *newSubmng(),}
+//	mos.queue.queue_init()
+//
+//	// 1. Make Publish Data
+//	pubLen := rand.Int()%100 + 1 // Set dataLength
+//	var pubDataList []MsgUnit
+//
+//	for i := 0; i < pubLen; i++{
+//		if i % 2 == 1 {
+//			msg := makePubData(true)
+//			mos.queue.push(msg)
+//			pubDataList = append(pubDataList, msg)
+//		} else{
+//			msg := makePubData(false)
+//			mos.queue.push(msg)
+//			pubDataList = append(pubDataList, msg)
+//		}
+//	}
+//
+//	// 2. Creates subscription data with (the same <Topic, Value> or (the same <Topic, difValue>
+//	//    And Add Subscription
+//	subLen := pubLen
+//	var subDataList []MsgUnit
+//	for i := 0; i < subLen; i++{
+//		Topic := pubDataList[i].(*PublishMsg).Topic
+//		Value := pubDataList[i].(*PublishMsg).Value
+//		if i % 2 == 1{
+//			msg := makeSubData(true, Topic, Value)
+//			mos.sub_mng.addSubscription(msg)
+//			subDataList = append(subDataList, msg)
+//
+//		} else{
+//			msg := makeSubData(false, Topic, Value)
+//			mos.sub_mng.addSubscription(msg)
+//			subDataList = append(subDataList, msg)
+//		}
+//	}
+//
+//	// 3. Watch Data
+//
+//  	// fmt.Println("PubData")
+//	// watchData(pubDataList, pubLen, false)
+//	// fmt.Println("\nSubData")
+//	// watchData(subDataList, subLen, true)
+//
+//	// 4. Matching
+//	for i := 0; i < pubLen; i++ {
+//		matching, pubmsg, err := mos.MatchingManager.Matching(&mos.queue, &mos.sub_mng)
+//
+//		// Check if matching is correct
+//		assert.Equal(t, nil, err)
+//
+//		fmt.Println("matching list = ", matching)
+//		fmt.Println("pub msg = ", pubmsg)
+//		fmt.Println("err ?= ", err)
+//	}
+//}

src/broker/modules/message.go

@@ -0,0 +1,152 @@
+package modules
+
+import "encoding/json"
+
+//*****메세지 타입 상수화
+const (
+	KGM = 1 + iota //KeyGenMessage
+	KSM            //KeyShareMessage
+	PM             //PublishMessage
+	SM             //SubscriptionMessage
+	RM             //RegisterMessage
+	WM             //WithdrawMessage
+)
+
+//*****메세지 틀*****
+type Message struct {
+	From    string //메세지 만든 주체의 Ip주소
+	Version string //메세지 버전
+	Time    string //메세지 만든 시간
+	Kind    int    //메세지 종류
+}
+
+type MsgUnit interface {
+	ConvertToJson() ([]byte, error) //메세지를 JSON형식으로 바꿔주는 멤버함수
+	CheckType() int                 //메세지의 종류를 반환하는 함수
+}
+
+//*****각 메세지 형식 및 정의**********
+
+//KeyGen 명령 메세지
+type PublishedImage struct {
+	Message
+	Topic []float64
+}
+
+type KeyGenMsg struct {
+	Message
+	iptable []string
+}
+
+//Key공유 메세지
+type KeyShareMsg struct {
+	Message
+	key string
+}
+
+//전달할 내용을 담은 메세지
+type PublishMsg struct {
+	Message
+	Topic   []int64 //대주제	ex)soccer
+	Value   []int64 //topic의 세부적인 내용 ex)ManCity or 40
+	Content []int64 //내용 ex)오늘 케빈 데 브라위너가 골을 넣었다
+}
+
+type SubscriptionImage struct {
+	Message
+	Topic []float64
+}
+
+//구독 정보를 담은 메세지
+type SubscriptionMsg struct {
+	Message
+	Topic    []int64  //대주제 ex)soccer
+	Value    []int64  //피연산자 ex)Mancity or 20 40
+	Operator []string //연산자	ex) ==, < && >
+	IsAlpha  bool     //value가 숫자인지 문자열인지, 문자열이면 단순비교, 숫자이면 범위연산
+}
+
+type RegisterImgMsg struct {
+	Message
+	PrivateKey float64
+}
+
+//Microservice 등록 메세지
+type RegisterMsg struct {
+	Message
+	PrivateKey int64
+}
+
+//Microservice 탈퇴 메세지(없앰)
+type WithdrawMsg struct {
+	Message
+}
+
+func (msg SubscriptionImage) ConvertToJson() ([]byte, error) {
+	js := msg
+	jsonBytes, err := json.Marshal(js)
+	return jsonBytes, err
+}
+
+//******ConverToJson을 메세지 종류별로 실행가능하게 구현******
+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 PublishedImage) 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 RegisterImgMsg) 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
+}
+
+//CheckType함수 구현
+func (msg Message) CheckType() int {
+	return msg.Kind //메세지 멤버변수 Kind 리턴
+}
+
+//KeyGenMsg 생성자
+func NewKeyGenMsg(table *MStable) *KeyGenMsg {
+	m := &KeyGenMsg{}
+	for _, value := range table.NodeTable { // MicroService테이블에서 ip주소를 다 가져와서 iptable에 넣음
+		m.iptable = append(m.iptable, value.GetIpaddr())
+	}
+	return m
+}

src/broker/modules/queue.go

@@ -0,0 +1,55 @@
+package modules
+
+import (
+	"errors"
+)
+
+type MsgQueue struct {
+	queue     chan MsgUnit
+	QueueFunc QueueOperate
+}
+
+type QueueOperate interface {
+	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.")
+		logger.Debug("queue is initialized")
+		return nil
+	} else {
+		close(mq.queue)
+		mq.queue = make(chan MsgUnit)
+		return nil
+	}
+}
+
+//push 함수
+func (mq *MsgQueue) push(msg MsgUnit) bool {
+	mq.queue <- msg
+	return true
+}
+
+//pop 함수
+func (mq *MsgQueue) pop(block bool) MsgUnit {
+	if block == false {
+		if len(mq.queue) == 0 {
+			return nil
+		} else {
+			return <-mq.queue
+		}
+	} else {
+		//queue에 데이터가 들어올 때까지 block
+		return <-mq.queue
+	}
+}

src/broker/modules/secure.go

@@ -0,0 +1,102 @@
+package modules
+
+import (
+	"fmt"
+	"strconv"
+)
+
+// 키관리 부분, 노드 입력받고 키 반환하는 부분 구현
+type Security struct {
+	KeyMap map[string]string
+}
+
+//Security 생성자
+func NewSecurity() *Security {
+	logger := NewMyLogger()
+	defer logger.Sync()
+	security := &Security{map[string]string{}}
+	defer logger.Debug("security setting complete.")
+	return security
+}
+
+type SecurityManager interface {
+	RegKey(rm RegisterMsg) // 원래 RegisterMsg
+	GetNodeKey(nodeName string) int64
+	ReEncrypt(fromKey int64, toKey int64, target []int64) []int64
+	ReEncPubMsg(fromPubMsg PublishMsg, nodeName string) PublishMsg
+	RemoveSecureKey(nodeName string) bool
+	//CompareTopic(topic1 []int64, topic2 []int64) int
+	//CompareDigit(topic1 int64, topic2 int64) int
+	//CompareAlpha(topic1 []int64, topic2 []int64) int
+}
+
+/*
+keyShareMsg 에서 각 노드의 private 키를 받아 keyMap 에 저장
+*/
+func (sc Security) RegKey(rm RegisterMsg) {
+	sc.KeyMap[rm.Message.From] = strconv.FormatInt(rm.PrivateKey, 10)
+}
+
+/**
+각 노드의 키를 주소를 이용하여 맵에서 가져옴
+*/
+func (sc Security) GetNodeKey(nodeName string) int64 {
+
+	messageStringKey := sc.KeyMap[nodeName]
+	mKey, err := strconv.ParseInt(messageStringKey, 10, 64)
+	if err != nil {
+		fmt.Println("GetNodeKey Error: key string to int64 parsing error.")
+	}
+	return mKey
+}
+
+/*
+reEncrypt 해서 슬라이스 반환
+*/
+func (sc Security) ReEncrypt(fromKey int64, toKey int64, target []int64) []int64 {
+	var tmpTarget []int64
+	for index := range target {
+		tmpTarget = append(tmpTarget, target[index]-fromKey+toKey)
+	}
+
+	return tmpTarget
+}
+
+func (sc Security) ReEncryptWithoutPrivateKey(toKey int64, target []int64) []int64 {
+	var tmpTarget []int64
+	for index := range target {
+		tmpTarget = append(tmpTarget, target[index]+toKey)
+	}
+
+	return tmpTarget
+}
+
+// topic과 value는 m+k로만 존재하므로 ReEnc과정에서 subscriber의 개인키만 더해주면 된다.
+func (sc Security) ReEncPubMsg(fromPubMsg PublishMsg, nodeName string) PublishMsg {
+	toKey := sc.GetNodeKey(nodeName)
+	fromKey := sc.GetNodeKey(fromPubMsg.Message.From)
+
+	toPubMsg := PublishMsg{}
+	toPubMsg.Message = fromPubMsg.Message
+	toPubMsg.Topic = sc.ReEncryptWithoutPrivateKey(toKey, fromPubMsg.Topic)
+	toPubMsg.Value = sc.ReEncryptWithoutPrivateKey(toKey, fromPubMsg.Value)
+	toPubMsg.Content = sc.ReEncrypt(fromKey, toKey, fromPubMsg.Content)
+
+	return toPubMsg
+}
+
+//Key제거 함수
+func (sc *Security) RemoveSecureKey(nodeName string) bool {
+	logger := NewMyLogger()
+	defer logger.Sync()
+	//삭제 전 존재여부 확인
+	_, exists := sc.KeyMap[nodeName]
+
+	if exists {
+		delete(sc.KeyMap, nodeName)
+		logger.Debug("[" + nodeName + "] : delete Key successful")
+		return true
+	} else {
+		return false
+	}
+}

src/broker/modules/secure_test.go

@@ -0,0 +1,139 @@
+package modules
+
+import (
+	"fmt"
+)
+
+//func TestCompare(t *testing.T) {
+//	var security = NewSecurity()
+//	var sm SecurityManager
+//	sm = security
+//
+//	ksm := KeyShareMsg{Message: Message{From: "1.1.1.1", Version: "1", Time: "2", Kind: 1}, key: "1234"}
+//	sm.RegKey(ksm)
+//	sm.GetNodeKey(ksm.Message.From)
+//	fmt.Println(sm.GetNodeKey(ksm.Message.From))
+//	var targetKey []int64
+//	targetKey = []int64{1234, 1235, 1236}
+//	fmt.Println(sm.ReEncrypt(sm.GetNodeKey(ksm.Message.From), 0, targetKey))
+//	//fmt.Println(sm.CompareDigit(1236, 1234))
+//
+//}
+
+func CreatePubMsg(msg Message, topic string, value string, content string) *PublishMsg {
+	toPubMsg := new(PublishMsg)
+	toPubMsg.Message = msg
+
+	intArr := []rune(topic)
+	//fmt.Print("Topic length ")
+	//fmt.Println(len(intArr))
+	//fmt.Println(len(toPubMsg.Topic))
+	for index := 0; index < len(intArr); index++ {
+		toPubMsg.Topic = append(toPubMsg.Topic, int64(intArr[index]))
+	}
+	//fmt.Println(len(toPubMsg.Topic))
+	intArr = []rune(value)
+	for index := 0; index < len(intArr); index++ {
+		toPubMsg.Value = append(toPubMsg.Value, int64(intArr[index]))
+	}
+	intArr = []rune(content)
+	for index := 0; index < len(intArr); index++ {
+		toPubMsg.Content = append(toPubMsg.Content, int64(intArr[index]))
+	}
+
+	return toPubMsg
+}
+
+func EncryptionMsg(msg *PublishMsg, gyKey int64, privateKey int64) *PublishMsg {
+	for index := range msg.Topic {
+		msg.Topic[index] = msg.Topic[index] + gyKey
+	}
+	for index := range msg.Value {
+		msg.Value[index] = msg.Value[index] + gyKey
+	}
+	for index := range msg.Content {
+		msg.Content[index] = msg.Content[index] + gyKey + privateKey
+	}
+
+	return msg
+}
+
+func DecryptionMsg(msg *PublishMsg, gyKey int64, privateKey int64) {
+	for index := range msg.Topic {
+		msg.Topic[index] = msg.Topic[index] - gyKey - privateKey
+	}
+	for index := range msg.Value {
+		msg.Value[index] = msg.Value[index] - gyKey - privateKey
+	}
+	for index := range msg.Content {
+		msg.Content[index] = msg.Content[index] - gyKey - privateKey
+	}
+
+	var runeArr []rune
+	for index := range msg.Topic {
+		runeArr = append(runeArr, rune(int(msg.Topic[index])))
+	}
+	fmt.Println("Topic is: " + string(runeArr))
+	runeArr = nil
+
+	for index := range msg.Value {
+		runeArr = append(runeArr, rune(int(msg.Value[index])))
+	}
+	fmt.Println("Value is: " + string(runeArr))
+	runeArr = nil
+
+	for index := range msg.Content {
+		runeArr = append(runeArr, rune(int(msg.Content[index])))
+	}
+	fmt.Println("Content is: " + string(runeArr))
+	runeArr = nil
+
+}
+
+func printMsg(msg *PublishMsg) {
+	var runeArr []rune
+	for index := range msg.Topic {
+		runeArr = append(runeArr, rune(int(msg.Topic[index])))
+	}
+	fmt.Println("Topic is: " + string(runeArr))
+	runeArr = nil
+
+	for index := range msg.Value {
+		runeArr = append(runeArr, rune(int(msg.Value[index])))
+	}
+	fmt.Println("Value is: " + string(runeArr))
+	runeArr = nil
+
+	for index := range msg.Content {
+		runeArr = append(runeArr, rune(int(msg.Content[index])))
+	}
+	fmt.Println("Content is: " + string(runeArr) + "\n\n")
+	runeArr = nil
+}
+
+// From "1.1.1.1" to "3.3.3.3" node
+//func TestReEnc(t *testing.T) {
+//	var security = NewSecurity()
+//	var sm SecurityManager
+//	sm = security
+//	security.KeyMap["1.1.1.1"] = "56789"
+//	security.KeyMap["3.3.3.3"] = "99999"
+//
+//	//fmt.Println(sm.GetNodeKey("3.3.3.3"))
+//	msg := Message{From: "1.1.1.1", Version: "1", Time: "123", Kind: 3}
+//	//fmt.Println(msg)
+//	publishMsg := CreatePubMsg(msg, "soccer123한글", "playerList", "Son and 10 players")
+//	fmt.Println(publishMsg)
+//	fmt.Println("original publish message is...")
+//	printMsg(publishMsg)
+//	encPublishMsg := EncryptionMsg(publishMsg, 1234, 56789)
+//	fmt.Println("encrypt publish message by publisher's private key")
+//	printMsg(encPublishMsg)
+//	fmt.Println(encPublishMsg)
+//	reEncPublishMsg := sm.ReEncPubMsg(*encPublishMsg, "3.3.3.3")
+//	fmt.Println("re-encrypt publish message by subscriber's private key")
+//	printMsg(reEncPublishMsg)
+//	//fmt.Println(reEncPublishMsg)
+//	fmt.Println("decrypted publish message is...")
+//	DecryptionMsg(reEncPublishMsg, 1234, 99999)
+//}

src/broker/modules/subscription.go

@@ -0,0 +1,301 @@
+package modules
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+)
+
+// Structure for managing subscriptions
+type sub_manager struct {
+	list       topicList          // Data Structure for Manage TopicNode
+	count_sub  int                // Subscription#
+	emptylist  []int              // For administrate Subscription#(Deleted)
+	ip2sub     map[string][]int   // For mapping {ip : Sub#s List}
+	sub2ip     map[int]string     // For mapping {Sub# : ip}
+	sub2node   map[int][]nodeInfo // For mapping {Sub# : NodeInfo List}
+	israngesub map[int]bool       // To manage when deleted
+
+	subnum2ip   map[int]string
+	subnum2conv map[int][]float64
+}
+
+type nodeInfo struct {
+	valNodeList []*valueNode
+	topic       []int64
+}
+
+type convNode struct {
+	subNum int
+}
+
+func (manager *sub_manager) Initialize() {
+	// Some initialize
+	manager.ip2sub = make(map[string][]int)
+	manager.sub2ip = make(map[int]string)
+	manager.sub2node = make(map[int][]nodeInfo)
+	manager.israngesub = make(map[int]bool)
+
+	manager.subnum2ip = make(map[int]string)
+	manager.subnum2conv = make(map[int][]float64)
+}
+
+func newSubmng() *sub_manager {
+	subMng := &sub_manager{}
+	subMng.Initialize()
+	return subMng
+}
+
+func (manager *sub_manager) isDuplicated(msg MsgUnit) bool {
+	//from := msg.(SubscriptionImage).From
+	//topic := msg.(SubscriptionImage).Topic
+	//value := msg.(SubscriptionMsg).Value
+	//operator := msg.(SubscriptionMsg).Operator
+	//subList := manager.ip2sub[from]
+	canFind := false
+
+	//for i := 0; i < len(subList) && canFind == false; i++ {
+	//	sub := subList[i]
+	//	nodeinfoList := manager.sub2node[sub]
+	//
+	//	for j := 0; j < len(nodeinfoList); j++ {
+	//		node := nodeinfoList[j]
+
+	//if Compare(node.topic, topic) == 0 {
+	//	cnt := 0
+	//	if len(operator) == 1 {
+	//		valPtr := node.valNodeList[0]
+	//		if Compare(valPtr.val, value) == 0 {
+	//			op := operator[0]
+	//			cnt += valPtr.findOperatorList(op, sub, true)
+	//		}
+	//	} else {
+	//		leftop := operator[0]
+	//		logicalop := operator[1]
+	//		rightop := operator[2]
+	//
+	//		leftValuePtr := node.valNodeList[0]
+	//		rightValuePtr := node.valNodeList[1]
+	//
+	//		nodeValList := []int64{leftValuePtr.val[0], rightValuePtr.val[0]}
+	//
+	//		if Compare(nodeValList, value) == 0 {
+	//			if logicalop == "&&" {
+	//				cnt += leftValuePtr.findOperatorList(leftop, sub, false)
+	//				cnt += rightValuePtr.findOperatorList(rightop, sub, false)
+	//			} else {
+	//				cnt += leftValuePtr.findOperatorList(leftop, sub, true)
+	//				cnt += rightValuePtr.findOperatorList(rightop, sub, true)
+	//			}
+	//		}
+	//	}
+	//	if cnt == len(node.valNodeList) {
+	//		canFind = true
+	//		break
+	//	}
+	//}
+	//	}
+	//}
+	return canFind
+}
+
+// To Insert sub#
+func (manager *sub_manager) addSubscription(msg MsgUnit) error {
+	topic := msg.(SubscriptionImage).Topic
+	//value := msg.(SubscriptionMsg).Value
+	//operator := msg.(SubscriptionMsg).Operator
+	subnumber := 0
+
+	// 0. Check if same IP & same <Topic, Value> exists
+	if manager.isDuplicated(msg) == true {
+		return errors.New("Duplicater Subscription")
+	}
+
+	// 1. Mapping incoming IP address to sub #
+	if len(manager.emptylist) == 0 {
+		subnumber = manager.count_sub
+		manager.ip2sub[msg.(SubscriptionImage).From] = append(manager.ip2sub[msg.(SubscriptionImage).From], subnumber)
+		manager.sub2ip[subnumber] = msg.(SubscriptionImage).From
+
+		manager.subnum2conv[subnumber] = msg.(SubscriptionImage).Topic //conv용
+
+		manager.count_sub++
+
+	} else {
+		subnumber := manager.emptylist[len(manager.emptylist)-1]
+		manager.emptylist = manager.emptylist[:len(manager.emptylist)-1]
+		manager.ip2sub[msg.(SubscriptionImage).From] = append(manager.ip2sub[msg.(SubscriptionImage).From], subnumber)
+		manager.sub2ip[subnumber] = msg.(SubscriptionImage).From
+
+		manager.subnum2conv[subnumber] = msg.(SubscriptionImage).Topic //conv용
+	}
+
+	topicptr := manager.list.head
+	existTopic := false
+
+	// 2. Add Subscription
+
+	// Find topic in topiclist, add if not found
+	for topicptr != nil {
+		if ConvCompare(topicptr.topic, topic) == 0 {
+			existTopic = true
+			break
+		}
+		topicptr = topicptr.next
+	}
+
+	if !existTopic {
+		manager.list.addConvTopic(topic)
+		topicptr = manager.list.tail
+	}
+	fmt.Println("SAVED SUB SM ")
+	fmt.Println(FloatSlice2String(topic))
+
+	// 안씀 오류시 날려
+	//var addValNodeList []*valueNode
+
+	// if single expression
+	//if len(operator) == 1 {
+	//	valptr := topicptr.list.getValueNodePos(value)
+	//	op := operator[0]
+	//
+	//	if valptr == nil {
+	//		topicptr.list.addValueNode(value)
+	//		valptr = topicptr.list.tail
+	//	}
+	//
+	//	addValNodeList = append(addValNodeList, valptr)
+	//	manager.sub2node[subnumber] = append(manager.sub2node[subnumber], nodeInfo{addValNodeList, topic})
+	//	manager.israngesub[subnumber] = false
+	//	valptr.insertSub(op, subnumber, true)
+	//
+	//	return nil // AddSubscription ok
+	//} else {
+	//// if Multi expression
+	//// (ex) : { (234 < x) && (x <= 1293) } , { (234 < x) || (x < 1293) }
+	//
+	//leftOperator := operator[0]
+	//logicalOperator := operator[1] // For compound expressions bounded by '&&' and '||'
+	//rightOperator := operator[2]
+	//
+	//// Find ValueNode = (topiclist[topic].list.val == value)
+	//valptr1 := topicptr.list.getValueNodePos([]int64{value[0]})
+	//valptr2 := topicptr.list.getValueNodePos([]int64{value[1]})
+	//
+	//if valptr1 == nil {
+	//	topicptr.list.addValueNode([]int64{value[0]})
+	//	valptr1 = topicptr.list.tail
+	//}
+	//if valptr2 == nil {
+	//	topicptr.list.addValueNode([]int64{value[1]})
+	//	valptr2 = topicptr.list.tail
+	//}
+	//
+	//addValNodeList = append(addValNodeList, valptr1)
+	//addValNodeList = append(addValNodeList, valptr2)
+	//manager.sub2node[subnumber] = append(manager.sub2node[subnumber], nodeInfo{addValNodeList, topic})
+	//
+	//if logicalOperator == "&&" {
+	//	// If they are enclosed in '&&' -> Insert Value to range_operator_list
+	//	manager.israngesub[subnumber] = true
+	//	valptr1.insertSub(leftOperator, subnumber, false)
+	//	valptr2.insertSub(rightOperator, subnumber, false)
+	//
+	//} else {
+	//	// if they are enclosed in '||' -> Insert Value to single_operator_list
+	//	manager.israngesub[subnumber] = false
+	//	valptr1.insertSub(leftOperator, subnumber, true)
+	//	valptr2.insertSub(rightOperator, subnumber, true)
+	//}
+	//	return nil // addSubscription ok
+	//}
+	return nil
+	//return errors.New("Can't addSubscription")
+}
+
+// To delete subscriptions
+func (manager *sub_manager) delete(from string) error {
+	ip := from
+	cand := manager.ip2sub[ip]
+
+	for i := 0; i < len(cand); i++ {
+		sub := cand[i]
+
+		for j := 0; j < len(manager.sub2node[sub]); j++ {
+			nodeinfo := manager.sub2node[sub][j]
+			node := nodeinfo.valNodeList
+
+			if manager.israngesub[sub] {
+				for k := 0; k < len(node); k++ {
+					pos := findSub(node[k].range2sub_s, sub)
+					if pos != -1 {
+						node[k].range2sub_s = remove(node[k].range2sub_s, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].range2sub_es, sub)
+					if pos != -1 {
+						node[k].range2sub_es = remove(node[k].range2sub_es, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].range2sub_b, sub)
+					if pos != -1 {
+						node[k].range2sub_b = remove(node[k].range2sub_b, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].range2sub_eb, sub)
+					if pos != -1 {
+						node[k].range2sub_eb = remove(node[k].range2sub_eb, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+				}
+			} else {
+				for k := 0; k < len(node); k++ {
+					pos := findSub(node[k].single2sub_s, sub)
+					if pos != -1 {
+						node[k].single2sub_s = remove(node[k].single2sub_s, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].single2sub_es, sub)
+					if pos != -1 {
+						node[k].single2sub_es = remove(node[k].single2sub_es, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].single2sub_b, sub)
+					if pos != -1 {
+						node[k].single2sub_b = remove(node[k].single2sub_b, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].single2sub_eb, sub)
+					if pos != -1 {
+						node[k].single2sub_eb = remove(node[k].single2sub_eb, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+
+					pos = findSub(node[k].single2sub_e, sub)
+					if pos != -1 {
+						node[k].single2sub_e = remove(node[k].single2sub_e, pos)
+						manager.emptylist = append(manager.emptylist, sub)
+					}
+				}
+			}
+		}
+		manager.ip2sub[ip] = nil // Delete sub#s mapped to Ip address
+		return nil
+	}
+	return errors.New("Don't Exist Subscription to delete")
+}
+
+func FloatSlice2String(target []float64) string {
+	var targetString string
+	targetString = ""
+	for _, value := range target {
+		targetString += strconv.FormatFloat(float64(value), 'f', -1, 64) + " "
+	}
+	return targetString
+}

src/broker/modules/subscription_test.go

@@ -0,0 +1,289 @@
+package modules
+
+//
+//import (
+//	"fmt"
+//	_ "fmt"
+//	"github.com/stretchr/testify/assert"
+//	"math/rand"
+//	"strconv"
+//	"testing"
+//	"time"
+//)
+//
+//func makeData(IsAlpha bool) MsgUnit{
+//	rand.Seed(time.Now().UnixNano())
+//	// Set Ipaddr
+//	msg := Message{"", "1.0", "", SM}
+//	for i := 0; i < 4; i++{
+//		itoa := strconv.Itoa(rand.Int() % 256)
+//		msg.from += itoa
+//		if i != 3{
+//			msg.from += "."
+//		}
+//	}
+//
+//	// Set Time
+//	msg.time += strconv.Itoa(rand.Int()%24) + ":"
+//	msg.time += strconv.Itoa(rand.Int()%60)
+//
+//	// Set Topic, Value, Operator
+//	Topic := []int64{}
+//	Value := []int64{}
+//	Operator := []string{}
+//	candOp := []string{">", ">=", "<=" ,"<", "=="}
+//	logicalOp := []string{"&&", "||"}
+//
+//	if IsAlpha {
+//		topicLen := rand.Int() % 10 + 1
+//		for i := 0 ; i < topicLen; i++{
+//			Topic = append(Topic, rand.Int63())
+//		}
+//
+//		valueLen := rand.Int() % 10 + 1
+//		for i := 0 ; i < valueLen; i++{
+//			Value = append(Value, rand.Int63())
+//		}
+//
+//		Operator = append(Operator, "==")
+//
+//	} else{
+//		valueLen := rand.Int() % 2 + 1
+//
+//		Topic = append(Topic, rand.Int63())
+//		Value = append(Value, rand.Int63())
+//
+//		if valueLen == 1{
+//			Operator = append(Operator, candOp[rand.Int()%5])
+//		} else{
+//			Value = append(Value, rand.Int63())
+//			op := rand.Int()%2
+//			Operator = append(Operator, candOp[rand.Int()%2])
+//			if op == 0 {
+//				Operator = append(Operator, logicalOp[0])
+//			} else{
+//				Operator = append(Operator, logicalOp[1])
+//			}
+//			Operator = append(Operator, candOp[rand.Int()%2 + 2])
+//		}
+//	}
+//
+//	return &SubscriptionMsg{msg, Topic, Value, Operator, IsAlpha}
+//}
+//
+//func makeMsgList(dataLen int, IsAlpha bool) []MsgUnit{
+//	rand.Seed(time.Now().UnixNano())
+//	var ret []MsgUnit
+//	for i := 0 ; i < dataLen ; i++{
+//		ret = append(ret, makeData(IsAlpha))
+//	}
+//	return ret
+//}
+//
+//func checkOperatorList(isSingle bool, sub int, Operator string, l *valueNode) int{
+//	ret := -1
+//	if isSingle{
+//		switch Operator {
+//		case "<":
+//			ret = findSub(l.single2sub_s, sub)
+//		case "<=":
+//			ret = findSub(l.single2sub_es, sub)
+//		case ">":
+//			ret = findSub(l.single2sub_b, sub)
+//		case ">=":
+//			ret = findSub(l.single2sub_eb, sub)
+//		case "==":
+//			ret = findSub(l.single2sub_e, sub)
+//		}
+//
+//	} else{
+//		switch Operator {
+//		case "<":
+//			ret = findSub(l.range2sub_s, sub)
+//		case "<=":
+//			ret = findSub(l.range2sub_es, sub)
+//		case ">":
+//			ret = findSub(l.range2sub_b, sub)
+//		case ">=":
+//			ret = findSub(l.range2sub_eb, sub)
+//		}
+//	}
+//	return ret
+//}
+//
+//func watchData(msgList []MsgUnit, dataLen int, isSubscription bool){
+//	for i := 0; i < dataLen; i++{
+//		msg := msgList[i]
+//		if isSubscription {
+//			fmt.Println(
+//				"\nfrom = ", msg.(*SubscriptionMsg).Message.from,
+//				"\ntime = ", msg.(*SubscriptionMsg).Message.time,
+//				"\nTopic = ", msg.(*SubscriptionMsg).Topic,
+//				"\nValue = ", msg.(*SubscriptionMsg).Value,
+//				"\nOperator = ", msg.(*SubscriptionMsg).Operator,
+//				"\nIsAlpha ?= ", msg.(*SubscriptionMsg).IsAlpha,
+//			)
+//		} else{
+//			fmt.Println(
+//				"\nfrom = ", msg.(*PublishMsg).Message.from,
+//				"\ntime = ", msg.(*PublishMsg).Message.time,
+//				"\nTopic = ", msg.(*PublishMsg).Topic,
+//				"\nValue = ", msg.(*PublishMsg).Value,
+//			)
+//		}
+//	}
+//}
+//
+////Test addSubScription(1) (dif all [Topic, Value, Operator])
+//func Test_addSubscription_allDif(t *testing.T) {
+//	rand.Seed(time.Now().UnixNano())
+//
+//	// To Init sub_mng
+//	mos := Moscato{sub_mng: *newSubmng()}
+//
+//	// Make Data set(Subscription)
+//	var msgList []MsgUnit
+//	dataLen := 100
+//	msgList = makeMsgList(dataLen, false)
+//
+//	//Watch Data set
+//	//watchData(msgList, dataLen, true)
+//
+//	for i := 0; i < dataLen; i++ {
+//		msg := msgList[i]
+//		ip := msg.(*SubscriptionMsg).Message.from
+//		Topic := msg.(*SubscriptionMsg).Topic
+//		Value := msg.(*SubscriptionMsg).Value
+//		Operator := msg.(*SubscriptionMsg).Operator
+//		subnumber := mos.sub_mng.count_sub
+//		isSingle := true
+//
+//		// 0. Check addSubscription
+//		err := mos.sub_mng.addSubscription(msg)
+//		assert.Equal(t, nil, err)
+//
+//		// 1. Check if ip mapping is correct
+//		assert.Equal(t, subnumber, mos.sub_mng.ip2sub[ip][len(mos.sub_mng.ip2sub[ip])-1], "Ip mapping is failed")
+//
+//		// 2. Check topicNode
+//		topicPtr := mos.sub_mng.list.head
+//		for topicPtr != nil {
+//			if Compare(Topic, topicPtr.Topic) == 0 {
+//				break
+//			}
+//			topicPtr = topicPtr.next
+//		}
+//
+//		assert.Equal(t, Topic, topicPtr.Topic, "topicNode Add is failed")
+//
+//		// Check isSingle ?
+//		if len(Operator) == 3 && Operator[1] == "&&" {
+//			isSingle = false
+//		}
+//
+//		// 3. Check Value in ValueNode & Check Operator in ValueNode
+//		if !isSingle || (len(Operator) == 3 && Operator[1] == "||") {
+//			valptr1 := topicPtr.list.getValueNodePos([]int64{Value[0]})
+//			valptr2 := topicPtr.list.getValueNodePos([]int64{Value[1]})
+//
+//			assert.Equal(t, []int64{Value[0]}, valptr1.val)
+//			assert.Equal(t, []int64{Value[1]}, valptr2.val)
+//
+//			assert.NotEqual(t, -1, checkOperatorList(isSingle, subnumber, Operator[0], valptr1))
+//			assert.NotEqual(t, -1, checkOperatorList(isSingle, subnumber, Operator[2], valptr2))
+//
+//		} else {
+//			valptr := topicPtr.list.getValueNodePos(Value)
+//			assert.Equal(t, Value, valptr.val)
+//			assert.NotEqual(t, -1, checkOperatorList(isSingle, subnumber, Operator[0], valptr))
+//		}
+//	}
+//}
+//
+//// Test addSubScription(2) (same [Topic, Value] dif [Operator])
+//func Test_addSubscription_same_topicNvalue(t *testing.T) {
+//	rand.Seed(time.Now().UnixNano())
+//
+//	// To Init sub_mng
+//	mos := Moscato{sub_mng: *newSubmng()}
+//
+//	// Fix Topic & Value
+//	topicLen := rand.Int()%10 + 1
+//	staticTopic := make([]int64, topicLen)
+//	staticValue := []int64{rand.Int63()}
+//
+//
+//	// Make Data
+//	var msgList []MsgUnit
+//	dataLen := 100
+//	msgList = makeMsgList(dataLen, false)
+//
+//	for i := 0; i < topicLen; i++ {
+//		staticTopic[i] = rand.Int63()
+//	}
+//
+//	// Fix Same Topic & Value
+//	for i := 0; i < dataLen; i++ {
+//		Operator := msgList[i].(*SubscriptionMsg).Operator
+//		msgList[i].(*SubscriptionMsg).Topic = staticTopic
+//
+//		if len(Operator) == 1{
+//			msgList[i].(*SubscriptionMsg).Value = staticValue
+//		}
+//	}
+//
+//	// Watch Data set
+//	//watchData(msgList, dataLen, true)
+//
+//	for i := 0; i < dataLen; i++ {
+//		msg := msgList[i]
+//		err := mos.sub_mng.addSubscription(msg)
+//		assert.Equal(t, nil, err)
+//	}
+//
+//	topicPtr := mos.sub_mng.list.head
+//	for topicPtr != nil {
+//		if Compare(topicPtr.Topic, staticTopic) == 0 {
+//			break
+//		}
+//		topicPtr = topicPtr.next
+//	}
+//	watchValueNode(topicPtr.list.head)
+//
+//}
+//
+//func watchValueNode(ptr *valueNode) {
+//	valPtr := ptr
+//	for valPtr != nil{
+//		fmt.Println("Value = ", valPtr.val)
+//		if len(valPtr.single2sub_s) != 0 {
+//			fmt.Println("Single2sub (<) List = ", valPtr.single2sub_s)
+//		}
+//		if len(valPtr.single2sub_es) != 0 {
+//			fmt.Println("Single2sub (<=) List = ", valPtr.single2sub_es)
+//		}
+//		if len(valPtr.single2sub_b) != 0 {
+//			fmt.Println("Single2sub (>) List = ", valPtr.single2sub_b)
+//		}
+//		if len(valPtr.single2sub_eb) != 0 {
+//			fmt.Println("Single2sub (>=) List = ", valPtr.single2sub_eb)
+//		}
+//		if len(valPtr.single2sub_e) != 0 {
+//			fmt.Println("Single2sub (==) List = ", valPtr.single2sub_e)
+//		}
+//
+//		if len(valPtr.range2sub_s) != 0 {
+//			fmt.Println("range2sub (<) List = ", valPtr.range2sub_s)
+//		}
+//		if len(valPtr.range2sub_es) != 0 {
+//			fmt.Println("range2sub (<=) List = ", valPtr.range2sub_es)
+//		}
+//		if len(valPtr.range2sub_b) != 0 {
+//			fmt.Println("range2sub (>) List = ", valPtr.range2sub_b)
+//		}
+//		if len(valPtr.range2sub_eb) != 0 {
+//			fmt.Println("range2sub (>=) List = ", valPtr.range2sub_eb)
+//		}
+//		valPtr = valPtr.next
+//	}
+//}

src/github.com/apex/log

@@ -0,0 +1 @@
+Subproject commit 8da83152b5d6177b4bfe3d12810a5afd25355170

src/github.com/benbjohnson/clock

@@ -0,0 +1 @@
+Subproject commit 68df829297d4f02d5997a02f8c0e8d4b12f0b2a3

src/github.com/bmizerany/assert

@@ -0,0 +1 @@
+Subproject commit b7ed37b82869576c289d7d97fb2bbd8b64a0cb28

src/github.com/davecgh/go-spew

@@ -0,0 +1 @@
+Subproject commit d8f796af33cc11cb798c1aaeb27a4ebc5099927d

src/github.com/facebookgo/ensure

@@ -0,0 +1 @@
+Subproject commit 63f1cf65ac4cf1f4bd7e93292149a456001b0e0b

src/github.com/facebookgo/inject

@@ -0,0 +1 @@
+Subproject commit f23751cae28bef101a38b79266785d3b574cf7e8

src/github.com/facebookgo/stack

@@ -0,0 +1 @@
+Subproject commit 751773369052141c013c6e827a71e8f35c07879c

src/github.com/facebookgo/structtag

@@ -0,0 +1 @@
+Subproject commit 217e25fb96916cc60332e399c9aa63f5c422ceed

src/github.com/facebookgo/subset

@@ -0,0 +1 @@
+Subproject commit c811ad88dec4edb3d7af0a88b34e6865d7460ba2

src/github.com/go-kit/kit

@@ -0,0 +1 @@
+Subproject commit b34471ec558cb53ef29ca37cb4f93525d8fd6136

src/github.com/go-kit/log

@@ -0,0 +1 @@
+Subproject commit 71fa7d7d64785c98d3814e53b3544fd8cccff897

src/github.com/go-logfmt/logfmt

@@ -0,0 +1 @@
+Subproject commit 2fe45f2cf057d707c50bc1949e25ec2cd6b7a015

src/github.com/go-stack/stack

@@ -0,0 +1 @@
+Subproject commit 93c7c7e3550c72bc91dead1452a0020142e2a902

src/github.com/juliangruber/go-intersect

@@ -0,0 +1 @@
+Subproject commit 2e99d8c0a75f6975a52f7efeb81926a19b221214

src/github.com/kr/pretty

@@ -0,0 +1 @@
+Subproject commit 0c671bcf5d4de20095c1ced9e581466bb5cc035d

src/github.com/kr/text

@@ -0,0 +1 @@
+Subproject commit a988ae8c4d852fa9930b96d707cc0fc6b43c0787

src/github.com/mattn/go-colorable

@@ -0,0 +1 @@
+Subproject commit e1bb79c8d53c38a60962ad4b8f658226cc983710

src/github.com/mattn/go-isatty

@@ -0,0 +1 @@
+Subproject commit cdb00f17c375f76e4d01572d16de4eb14b48576a

src/github.com/pkg/errors

@@ -0,0 +1 @@
+Subproject commit 5dd12d0cfe7f152f80558d591504ce685299311e

src/github.com/pmezard/go-difflib

@@ -0,0 +1 @@
+Subproject commit 5d4384ee4fb2527b0a1256a821ebfc92f91efefc

src/github.com/rogpeppe/go-internal

@@ -0,0 +1 @@
+Subproject commit 115ce09fd6b421993aafa8cdae0171429e9bd2c4

src/github.com/rs/zerolog

@@ -0,0 +1 @@
+Subproject commit c0c2e11fc3cd04ae28d856789cf58ffd1666bc3f

src/github.com/sirupsen/logrus

@@ -0,0 +1 @@
+Subproject commit 79c5ab66aa2ce7d9ff7b3c437ebc22fcc519a967

src/github.com/stretchr/testify

@@ -0,0 +1 @@
+Subproject commit ab6dc3262822ed562480c19876b0257ace761e3e

src/go.uber.org/atomic

@@ -0,0 +1 @@
+Subproject commit 3504dfaa1fa414923b1c8693f45d2f6931daf229

src/go.uber.org/goleak

@@ -0,0 +1 @@
+Subproject commit 421b885faf413a221d1a86a17684fb3c45bd35c5

src/go.uber.org/multierr

@@ -0,0 +1 @@
+Subproject commit d49c2ba57443d9ddb09d5716b7c971f8ecbdce24

src/go.uber.org/zap

@@ -0,0 +1 @@
+Subproject commit 369c1bdf2a994a436ea89247432544ad7184020b

src/golang.org/x/sys

@@ -0,0 +1 @@
+Subproject commit 1d35b9e2eb4edf581781c7f3e2a36fac701f0a24

src/google.golang.org/grpc

@@ -0,0 +1 @@
+Subproject commit fbaf7c55821070944bb0ce342ba3c54cc521c6fe

src/gopkg.in/inconshreveable/log15.v2

@@ -0,0 +1 @@
+Subproject commit b30bc20e4fd12cec79a9aae62e91cfcf458bd253

src/gopkg.in/yaml.v2

@@ -0,0 +1 @@
+Subproject commit 7649d4548cb53a614db133b2a8ac1f31859dda8c

src/gopkg.in/yaml.v3

@@ -0,0 +1 @@
+Subproject commit 496545a6307b2a7d7a710fd516e5e16e8ab62dbc