chtwrs/chirpnest
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

test

Browse Source
This commit is contained in:
shoopea 2019-06-28 14:46:56 +08:00
parent 6a69fc034d
commit 6f95de1b78
1 changed files with 90 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

179
workers.go
View File

@ -31,7 +31,7 @@ func MQGetMsgWorker(id int, msgs chan<- ChatWarsMessage) {
time.Sleep(15)
err = s.Open()
}
log.Printf("MQGetMsgWorker[" + strconv.Itoa(id) + "] : Connected to RabbitMQ")
//log.Printf("MQGetMsgWorker[" + strconv.Itoa(id) + "] : Connected to RabbitMQ")
log.Printf("MQGetMsgWorker["+strconv.Itoa(id)+"] : Session address : %p.\n", &s)
/*
m, err := s.MQChannel.Consume(
@ -59,6 +59,95 @@ func MQGetMsgWorker(id int, msgs chan<- ChatWarsMessage) {
}
func MQKeepAliveWorker() {
s := MQSession{
User: cfg.Rabbit.User,
Password: cfg.Rabbit.Password,
Host: cfg.Rabbit.Host,
Path: cfg.Rabbit.Path,
SSL: false,
Queue: "keepalive",
isConnected: false,
}
//log.Printf("MQKeepAliveWorker : Starting.")
for true {
err := s.Open()
for err != nil {
logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
time.Sleep(15 * time.Second)
err = s.Open()
}
//log.Printf("MQKeepAliveWorker : Connected to RabbitMQ")
log.Printf("MQKeepAliveWorker : Session address : %p.\n", &s)
/*
m, err := s.MQChannel.Consume(
s.MQQueue.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
*/
for d := range s.MQDelivery {
log.Printf("MQKeepAliveWorker : Received a message: %s", string(d.Body))
x := MQKeepAlive{}
err = json.Unmarshal(d.Body, &x)
logOnError(err, "MQKeepAliveWorker : Can't unmarshal.\n"+string(d.Body))
if err == nil {
if x.Date.Add(10 * time.Second).Before(time.Now()) {
// outdated keep-alive coming from client
} else if v, ok := clientsKeepAlive.Load(x.UserID64); ok {
k := v.(*MQKeepAlive)
k.Date = x.Date
} else {
cs := MQSession{
User: cfg.Rabbit.User,
Password: cfg.Rabbit.Password,
Host: cfg.Rabbit.Host,
Path: x.Queue,
SSL: false,
Queue: "msg",
isConnected: false,
}
err = cs.Open()
logOnError(err, "MQKeepAliveWorker : Failed to open MQ session")
clientsKeepAlive.Store(x.UserID64, &x)
clientsQueue[x.UserID64] = &cs
c := TGCommand{
Type: commandSendMsg,
ToUserID64: x.UserID64,
Text: "Your client is connected.",
}
TGCmdQueue <- c
c = TGCommand{
Type: commandSendMsg,
ToUserID64: cfg.Bot.Admin,
Text: fmt.Sprintf("Client `%s` is connected.", x.Nickname),
}
TGCmdQueue <- c
clientSendCWMsg(x.UserID64, `🏅Me`)
/*
c = TGCommand{
Type: commandSendMsg,
FromUserID64: x.UserID64,
ToChatID64: userID64ChtWrsBot,
Text: `/hero`,
}
MQTGCmdQueue <- c
*/
}
}
}
}
log.Printf("MQKeepAliveWorker : Closing.")
}
func SQLCWMsgWorker(id int, msgs <-chan ChatWarsMessage, objIds chan<- int64) {
//log.Printf("SQLCWMsgWorker[" + strconv.Itoa(id) + "] : Starting.")
for m := range msgs {
@ -345,94 +434,6 @@ func MQTGCmdWorker(id int, cmds <-chan TGCommand) {
}
func MQKeepAliveWorker() {
s := MQSession{
User: cfg.Rabbit.User,
Password: cfg.Rabbit.Password,
Host: cfg.Rabbit.Host,
Path: cfg.Rabbit.Path,
SSL: false,
Queue: "keepalive",
isConnected: false,
}
//log.Printf("MQKeepAliveWorker : Starting.")
for true {
err := s.Open()
for err != nil {
logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
time.Sleep(15 * time.Second)
err = s.Open()
}
log.Printf("MQKeepAliveWorker : Connected to RabbitMQ")
/*
m, err := s.MQChannel.Consume(
s.MQQueue.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
*/
for d := range s.MQDelivery {
log.Printf("MQKeepAliveWorker : Received a message: %s", string(d.Body))
x := MQKeepAlive{}
err = json.Unmarshal(d.Body, &x)
logOnError(err, "MQKeepAliveWorker : Can't unmarshal.\n"+string(d.Body))
if err == nil {
if x.Date.Add(10 * time.Second).Before(time.Now()) {
// outdated keep-alive coming from client
} else if v, ok := clientsKeepAlive.Load(x.UserID64); ok {
k := v.(*MQKeepAlive)
k.Date = x.Date
} else {
cs := MQSession{
User: cfg.Rabbit.User,
Password: cfg.Rabbit.Password,
Host: cfg.Rabbit.Host,
Path: x.Queue,
SSL: false,
Queue: "msg",
isConnected: false,
}
err = cs.Open()
logOnError(err, "MQKeepAliveWorker : Failed to open MQ session")
clientsKeepAlive.Store(x.UserID64, &x)
clientsQueue[x.UserID64] = &cs
c := TGCommand{
Type: commandSendMsg,
ToUserID64: x.UserID64,
Text: "Your client is connected.",
}
TGCmdQueue <- c
c = TGCommand{
Type: commandSendMsg,
ToUserID64: cfg.Bot.Admin,
Text: fmt.Sprintf("Client `%s` is connected.", x.Nickname),
}
TGCmdQueue <- c
clientSendCWMsg(x.UserID64, `🏅Me`)
/*
c = TGCommand{
Type: commandSendMsg,
FromUserID64: x.UserID64,
ToChatID64: userID64ChtWrsBot,
Text: `/hero`,
}
MQTGCmdQueue <- c
*/
}
}
}
}
log.Printf("MQKeepAliveWorker : Closing.")
}
func MQTidyKeepAliveWorker() {
//log.Printf("MQTidyKeepAliveWorker : Starting.")
for true {