Commit 1e4a77d4 authored by Peter Cheng's avatar Peter Cheng

1.修正Error觸發後動作

2.修正工作佇列排程內容
parent 5e87fee0
package controllers
import (
"fmt"
"log"
"github.com/bitly/go-nsq"
......@@ -18,7 +17,7 @@ func postMessage(channel string, message string) {
w, _ := nsq.NewProducer(cfg.Message.Post.Address, config)
err := w.Publish(channel, []byte(message))
if err != nil {
fmt.Println(err)
log.Panicln(err)
}
w.Stop()
}
......@@ -27,6 +26,6 @@ func postMessage(channel string, message string) {
func getParams(c *gin.Context, params interface{}) {
err := c.BindJSON(params)
if err != nil {
log.Println(err)
log.Panicln(err)
}
}
package env
import "time"
//Config 系統參數
type Config struct {
Env string
......@@ -17,6 +19,16 @@ type Config struct {
Housekeeper struct {
Host string
}
Works struct {
Worker []*Worker
}
}
//Worker 工作佇列
type Worker struct {
Topic string
Channel string
Interval time.Duration
}
var cfg = &Config{}
......@@ -29,6 +41,10 @@ func init() {
cfg.Message.Topic = "Mail"
cfg.Message.Channel = "SendMail"
cfg.Housekeeper.Host = "http://127.0.0.1:8806"
workers := []*Worker{
&Worker{Topic: "Mail", Channel: "SendMail", Interval: time.Second * 2},
&Worker{Topic: "PunchClock", Channel: "UploadDailyPunchclockData", Interval: time.Second * 10}}
cfg.Works.Worker = workers
}
//GetEnv 取得環境參數
......
......@@ -44,24 +44,22 @@ func messageService() {
//InitConsumer 初始化消費者
func InitConsumer(topic string, channel string, host string) bool {
upStream := make(chan int, 1)
go func() {
interval := time.Second * 10
mQuery(host, "PunchClock", "UploadDailyPunchclockData", interval)
}()
go func() {
interval := time.Second * 2
mQuery(host, "Mail", "SendMail", interval)
}()
for _, worker := range cfg.Works.Worker {
go func(worker *env.Worker) {
mQuery(host, worker.Topic, worker.Channel, worker.Interval)
}(worker)
}
<-upStream
return true
}
//mQuery 啟用單一消費者
func mQuery(host string, topic string, channel string, interval time.Duration) *nsq.Consumer {
config := nsq.NewConfig()
config.LookupdPollInterval = interval
query, err := nsq.NewConsumer(topic, channel, config)
if err != nil {
log.Panic(err)
log.Panicln(err)
}
query.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
chat := string(message.Body)
......@@ -71,16 +69,18 @@ func mQuery(host string, topic string, channel string, interval time.Duration) *
}))
err = query.ConnectToNSQLookupd(host)
if err != nil {
panic(err)
log.Panicln(err)
}
return query
}
//getQuete 取得工作內容
func getQuete(message string) []string {
quete := strings.Split(message, "</UseService>")
return quete
}
//runServices 運行系統服務
func runServices(quete []string) {
path := quete[0]
params := []byte(quete[1])
......@@ -88,6 +88,7 @@ func runServices(quete []string) {
postURL(url, params)
}
//postURL 將資料打到REST API
func postURL(url string, params []byte) {
request, err := http.NewRequest("POST", url, bytes.NewBuffer(params))
request.Header.Set("X-Custom-Header", "counter")
......@@ -95,7 +96,7 @@ func postURL(url string, params []byte) {
client := &http.Client{}
resp, err := client.Do(request)
if err != nil {
log.Panic(err)
log.Panicln(err)
}
defer resp.Body.Close()
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment