Merge pull request #38 from p4tin/LogMessages

Add Logging of Messaged received+yaml config+Fix Publish Bug (non-exists topic)

.gitignore

@@ -4,3 +4,5 @@ goaws
 goaws_linux_amd64
 
 venv/
+logs/
+*.log

common/misc.go → common/common.go

@@ -6,8 +6,14 @@ import (
 	"crypto/rand"
 	"crypto/md5"
 	"encoding/hex"
+	"io/ioutil"
+	"os"
+	"log"
 )
 
+var LogMessages bool
+var LogFile	string
+
 func NewUUID() (string, error) {
 	uuid := make([]byte, 16)
 	n, err := io.ReadFull(rand.Reader, uuid)
@@ -27,3 +33,16 @@ func GetMD5Hash(text string) string {
 	hasher.Write([]byte(text))
 	return hex.EncodeToString(hasher.Sum(nil))
 }
+
+func LogMessage(msg string) {
+	if _, err := os.Stat(LogFile); os.IsNotExist(err) {
+		_, err := os.Create("/tmp/dat2")
+		if err != nil {
+			log.Println("could not create log file:", LogFile)
+			return
+		}
+	}
+	if LogMessages == true {
+		ioutil.WriteFile(LogFile, []byte(msg), 0644)
+	}
+}

conf/config.go

@@ -12,24 +12,26 @@ import (
 )
 
 type EnvSubsciption struct {
-	QueueName string
-	Raw bool
+	QueueName 	string
+	Raw 		bool
 }
 
 type EnvTopic struct {
-	Name string
-	Subscriptions []EnvSubsciption
+	Name 		string
+	Subscriptions 	[]EnvSubsciption
 }
 
 type EnvQueue struct {
-	Name string
+	Name 		string
 }
 
 type Environment struct {
-	Host string
-	Port string
-	Topics []EnvTopic
-	Queues []EnvQueue
+	Host 		string
+	Port 		string
+	LogMessages 	bool
+	LogFile 	string
+	Topics 		[]EnvTopic
+	Queues 		[]EnvQueue
 }
 
 var envs map[string]Environment
@@ -58,6 +60,17 @@ func LoadYamlConfig(env string, portNumber string) string {
 		}
 	}
 
+
+	common.LogMessages = false
+	common.LogFile = "./goaws_messages.log"
+
+	if envs[env].LogMessages == true {
+		common.LogMessages = true
+		if envs[env].LogFile != "" {
+			common.LogFile = envs[env].LogFile
+		}
+	}
+
 	sqs.SyncQueues.Lock()
 	for _, queue := range envs[env].Queues {
 		queueUrl := "http://" + envs[env].Host + ":" + portNumber +"/queue/" + queue.Name

goaws.yaml

@@ -2,6 +2,8 @@ Local:                              # Environment name that can be passed on the
                                     #     (i.e.: ./goaws [Local | Dev]  -- defaults to 'Local')
   Host: localhost                   # hostname of the goaws system  (for docker-compose this is the tag name of the container)
   Port: 4100                        # port to listen on.
+  LogMessages: true                 # Log messages (true/false)
+  LogFile: ./goaws_messages.log  # Log filename (for message logging
   Queues:                           # List of queues to create at startup
     - Name: local-queue1            # Queue name
     - Name: local-queue2            # Queue name
@@ -17,6 +19,8 @@ Local:                              # Environment name that can be passed on the
 Dev:                                # Another environment
   Host: localhost
   Port: 4100
+  LogMessages: true
+  LogFile: ./goaws_messages.log
   Queues:
     - Name: dev-queue1
     - Name: dev-queue2

gosns/gosns.go

@@ -239,26 +239,32 @@ func Publish(w http.ResponseWriter, req *http.Request) {
 	uriSegments := strings.Split(topicArn, ":")
 	topicName := uriSegments[len(uriSegments)-1]
 
-	for _, subs := range SyncTopics.Topics[topicName].Subscriptions {
-		if subs.Protocol == "sqs" {
-			queueUrl := subs.EndPoint
-			uriSegments := strings.Split(queueUrl, "/")
-			queueName := uriSegments[len(uriSegments) - 1]
-
-			msg := sqs.Message{}
-			log.Println("Subscript:", subs.EndPoint, "Raw: ", subs.Raw)
-			if subs.Raw == false {
-				msg.MessageBody = CreateMessageBody(messageBody, topicArn)
-			} else {
-				msg.MessageBody = []byte(messageBody)
+	_, ok := SyncTopics.Topics[topicName];
+	if ok {
+		for _, subs := range SyncTopics.Topics[topicName].Subscriptions {
+			if subs.Protocol == "sqs" {
+				queueUrl := subs.EndPoint
+				uriSegments := strings.Split(queueUrl, "/")
+				queueName := uriSegments[len(uriSegments) - 1]
+
+				msg := sqs.Message{}
+				if subs.Raw == false {
+					msg.MessageBody = CreateMessageBody(messageBody, topicArn)
+				} else {
+					msg.MessageBody = []byte(messageBody)
+				}
+				msg.MD5OfMessageAttributes = common.GetMD5Hash("GoAws")
+				msg.MD5OfMessageBody = common.GetMD5Hash(messageBody)
+				msg.Uuid, _ = common.NewUUID()
+				sqs.SyncQueues.Lock()
+				sqs.SyncQueues.Queues[queueName].Messages = append(sqs.SyncQueues.Queues[queueName].Messages, msg)
+				sqs.SyncQueues.Unlock()
+				common.LogMessage(fmt.Sprintf("%s: Topic: %s(%s), Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), topicName, queueName, msg.MessageBody))
 			}
-			msg.MD5OfMessageAttributes = common.GetMD5Hash("GoAws")
-			msg.MD5OfMessageBody = common.GetMD5Hash(messageBody)
-			msg.Uuid, _ = common.NewUUID()
-			sqs.SyncQueues.Lock()
-			sqs.SyncQueues.Queues[queueName].Messages = append(sqs.SyncQueues.Queues[queueName].Messages, msg)
-			sqs.SyncQueues.Unlock()
 		}
+	} else {
+		createErrorResponse(w, req, "TopicNotFound")
+		return
 	}
 
 	//Create the response

gosqs/gosqs.go

@@ -128,6 +128,8 @@ func SendMessage(w http.ResponseWriter, req *http.Request) {
 	SyncQueues.Lock()
 	SyncQueues.Queues[queueName].Messages = append(SyncQueues.Queues[queueName].Messages, msg)
 	SyncQueues.Unlock()
+	common.LogMessage(fmt.Sprintf("%s: Queue: %s, Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody))
+
 
 	respStruct := SendMessageResponse{"http://queue.amazonaws.com/doc/2012-11-05/", SendMessageResult{MD5OfMessageAttributes: msg.MD5OfMessageAttributes, MD5OfMessageBody: msg.MD5OfMessageBody, MessageId: msg.Uuid}, ResponseMetadata{RequestId: "00000000-0000-0000-0000-000000000000"}}
 	enc := xml.NewEncoder(w)