package main

import (

    "net"

    "os"

    "github.com/gpmgo/gopm/modules/goconfig"

    "github.com/go-stomp/stomp"

    "time"

    "strconv"

    "log"

    "strings"

)

// 限制goroutine数量

var limitChan = make(chan bool, ) // Todo 从配置文件中读取

// 限制同时处理消息数量

var msgChan = make(chan string, ) // Todo 从配置文件中读取

var activeMqLimitedChan = make(chan bool, )

var activeMq *stomp.Conn

var activeQueue string

var host string

var port string

var connectTimes =

var udpAddress = "0.0.0.0"    // Todo 从配置文件中读取

var udpPort = ""    // Todo 从配置文件中读取

var logFilePath = "/var/log/syslog_server/"

var configFilePath = "./config.ini"

// UDP goroutine 实现并发读取UDP数据

func udpProcess(conn *net.UDPConn) {

    defer func() {

        if e := recover(); e != nil {

            // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

            logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

            logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

            defer logFile.Close()

            if err != nil {

                log.Fatalln("open log file error: ", err.Error())

            }

            logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

            // 记录错误日志

            logger.Println("udpProcess error:", e)

        }

        // 释放出一个协程

        <- limitChan

    }()

    // 最大读取数据大小

    data := make([]byte, )

    n, _, err := conn.ReadFromUDP(data)

    if err != nil {

        panic(err)

    }

    // 获取对端的IP地址

    // remoteAddr := conn.RemoteAddr()

    // msgChan <- remoteAddr.String() + " " + string(data[:n])

    msgChan <- string(data[:n])

}

func udpServer(address, port string) {

    // @todo 如何防止udpServer 一直Panic导致无限循环重启

    defer func() {

        if e := recover(); e != nil {

            // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

            logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

            logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

            defer logFile.Close()

            if err != nil {

                log.Fatalln("open log file error: ", err.Error())

            }

            logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

            // 记录错误日志

            logger.Println("udpServer error:", e)

            // udpServer启动失败后，间隔10秒后重试

            time.Sleep( * time.Second)

            udpServer(udpAddress, udpPort)

        }

    }()

    udpAddr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(address, port))

    conn, err := net.ListenUDP("udp", udpAddr)

    defer conn.Close()

    if err != nil {

        panic(err)

    }

    for {

        limitChan <- true

        go udpProcess(conn)

    }

}

// 读取ActiveMQ配置信息

func getConfiguration(){

    defer func() {

        if e := recover(); e != nil {

            // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

            logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

            logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

            defer logFile.Close()

            if err != nil {

                log.Fatalln("open log file error: ", err.Error(), ", programing exit.")

                os.Exit()

            }

            logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

            // 记录错误日志

            logger.Println("Get Configuration error:", e)

        }

    }()

    configFile, err := goconfig.LoadConfigFile(configFilePath)

    if err != nil {

        panic(err)

    }

    host, err = configFile.GetValue("active_mq", "host")

    if err != nil {

        // 如果没有配置主机，则使用本地主机

        host = "127.0.0.1"

    }

    port, err = configFile.GetValue("active_mq", "port")

    if err != nil {

        // 如果没配置端口，则使用默认端口

        port = ""

    }

    activeQueue, err = configFile.GetValue("active_mq", "queue")

    if err != nil {

        // 如果没配置端口，则使用默认队列名

        activeQueue = "syslog.queue"

    }

}

// 使用IP和端口连接到ActiveMQ服务器, 返回ActiveMQ连接对象

func connActiveMq(){

    // @todo 如何防止无限循环

    defer func() {

        if e := recover(); e != nil {

            // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

            logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

            logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

            defer logFile.Close()

            if err != nil {

                log.Fatalln("open log file error: ", err.Error())

            }

            logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

            // 记录错误日志

            logger.Println("connActiveMq error:", e)

            // ActiveMQ服务器连接失败后，间隔3秒后重试

            time.Sleep( * time.Second)

            activeMq = nil

            connActiveMq()

        }

    }()

    // @todo 实现断开重连

    if activeMq == nil {

        var err error

        activeMq, err = stomp.Dial("tcp", net.JoinHostPort(host, port))

        if err != nil {

            connectTimes ++

            if connectTimes >=  {

                time.Sleep( * time.Second)

            }else if connectTimes >=  {

                time.Sleep( * time.Second)

            }else {

                time.Sleep( * time.Second)

            }

            panic(err.Error() + ", 重新连接ActiveMQ, 已重试次数: " + strconv.Itoa(connectTimes))

        }else {

            connectTimes =

        }

    }

}

func activeMqProducer(c chan string){

    // @todo 如何防止activeMqProducer 退出

    defer func() {

        if e := recover(); e != nil {

            // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

            logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

            logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

            defer logFile.Close()

            if err != nil {

                log.Fatalln("open log file error: ", err.Error())

            }

            logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

            // 记录错误日志

            logger.Println("activeMqProducer error:", e)

            // 重试

            go activeMqProducer(msgChan)

        }

    }()

    for{

        activeMqLimitedChan <- true    // 限制开启协程数量

        contentMsg := <-c

        go func() {

            defer func() {

                if e := recover(); e != nil {

                    err := os.MkdirAll(logFilePath, )

                    log.Fatalln("create log dirctory error: ", err.Error())

                    // 初始化日志，每天生成一个日志文件，日志文件名以日志结尾

                    logFileName := logFilePath + "server-" + strings.Split(time.Now().String(), " ")[] + ".log"

                    logFile, err := os.OpenFile(logFileName, os.O_RDWR|os.O_CREATE|os.O_APPEND, ) // 应该判断error，此处简略

                    defer logFile.Close()

                    if err != nil {

                        log.Fatalln("open log file error: ", err.Error())

                    }

                    logger := log.New(logFile, "[Error] ", log.Ldate|log.Ltime|log.Lshortfile)

                    // 记录错误日志

                    logger.Println("activeMqProducer error:", e)

                }

                // 释放出一个协程

                <- activeMqLimitedChan

            }()

            err := activeMq.Send(activeQueue, "text/plain", []byte(contentMsg))

            if err != nil {

                if err.Error() == "connection already closed"{

                    activeMq = nil

                    connActiveMq()

                    activeMq.Send(activeQueue, "text/plain", []byte(contentMsg))

                }

                panic(err)

            }

        }()

    }

}

func init(){

    // 初始化 ActiveMQ 配置

    getConfiguration()

    // 连接到 ActiveMQ 服务器

    connActiveMq()

    // 启动一个协程将Syslog消息放入ActiveMQ队列中

    go activeMqProducer(msgChan)

}

func main() {

    defer activeMq.Disconnect()

    udpServer(udpAddress, udpPort)

}