This commit is contained in:
shoopea 2019-06-28 17:30:49 +08:00
parent cc8569b765
commit 836bc9bf1b

View File

@ -94,6 +94,124 @@ func MQGetMsgWorker(id int, msgs chan<- ChatWarsMessage) {
} }
func MQKeepAliveWorker() {
//log.Printf("MQKeepAliveWorker : Starting.")
var err error
c := MQClient{
User: cfg.Rabbit.User,
Password: cfg.Rabbit.Password,
Host: cfg.Rabbit.Host,
Path: cfg.Rabbit.Path,
SSL: false,
}
for true {
c.Connection, err = amqp.Dial("amqp://" + c.User + ":" + c.Password + "@" + c.Host + "/" + c.Path)
logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Cannot open MQ connection")
if err != nil {
c.Connection.Close()
time.Sleep(15 * time.Second)
continue
}
c.Channel, err = c.Connection.Channel()
logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Cannot open MQ channel")
if err != nil {
c.Channel.Close()
c.Connection.Close()
time.Sleep(15 * time.Second)
continue
}
c.Queue, err = c.Channel.QueueDeclare(
"keepalive", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to declare a queue")
if err != nil {
c.Channel.Close()
c.Connection.Close()
time.Sleep(15 * time.Second)
continue
}
m, err := c.Channel.Consume(
c.Queue.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to register a consumer")
if err != nil {
c.Channel.Close()
c.Connection.Close()
time.Sleep(15 * time.Second)
continue
}
for d := range m {
// 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 {
clt := MQClient{}
clt.Connection, err = amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + x.Queue)
logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
clt.Channel, err = clt.Connection.Channel()
logOnError(err, "MQKeepAliveWorker : Failed to open a channel")
clt.Queue, err = clt.Channel.QueueDeclare(
"msg", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
logOnError(err, "MQKeepAliveWorker : Failed to declare a queue")
clientsKeepAlive.Store(x.UserID64, &x)
clientsQueue[x.UserID64] = &clt
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`)
clientSendCWMsg(x.UserID64, `/hero`)
}
}
}
c.Channel.Close()
c.Connection.Close()
}
log.Printf("MQKeepAliveWorker : Closing.")
}
func SQLCWMsgWorker(id int, msgs <-chan ChatWarsMessage, objIds chan<- int64) { func SQLCWMsgWorker(id int, msgs <-chan ChatWarsMessage, objIds chan<- int64) {
//log.Printf("SQLCWMsgWorker[" + strconv.Itoa(id) + "] : Starting.") //log.Printf("SQLCWMsgWorker[" + strconv.Itoa(id) + "] : Starting.")
for m := range msgs { for m := range msgs {
@ -381,119 +499,6 @@ func MQTGCmdWorker(id int, cmds <-chan TGCommand) {
} }
func MQKeepAliveWorker() {
//log.Printf("MQKeepAliveWorker : Starting.")
for true {
conn, err := amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + cfg.Rabbit.Path)
logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
if err != nil {
conn.Close()
time.Sleep(15 * time.Second)
} else {
ch, err := conn.Channel()
logOnError(err, "MQKeepAliveWorker : Failed to open a channel")
if err != nil {
ch.Close()
conn.Close()
time.Sleep(15 * time.Second)
} else {
q, err := ch.QueueDeclare(
"keepalive", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
logOnError(err, "MQKeepAliveWorker : Failed to declare a queue")
if err != nil {
ch.Close()
conn.Close()
time.Sleep(15 * time.Second)
} else {
m, err := ch.Consume(
q.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
logOnError(err, "MQKeepAliveWorker : Failed to register a consumer")
if err != nil {
ch.Close()
conn.Close()
time.Sleep(15 * time.Second)
} else {
for d := range m {
// 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 {
clt := MQClient{}
clt.Connection, err = amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + x.Queue)
logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
clt.Channel, err = clt.Connection.Channel()
logOnError(err, "MQKeepAliveWorker : Failed to open a channel")
clt.Queue, err = clt.Channel.QueueDeclare(
"msg", // name
false, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
logOnError(err, "MQKeepAliveWorker : Failed to declare a queue")
clientsKeepAlive.Store(x.UserID64, &x)
clientsQueue[x.UserID64] = &clt
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
*/
}
}
}
ch.Close()
conn.Close()
}
}
}
}
}
log.Printf("MQKeepAliveWorker : Closing.")
}
func MQTidyKeepAliveWorker() { func MQTidyKeepAliveWorker() {
//log.Printf("MQTidyKeepAliveWorker : Starting.") //log.Printf("MQTidyKeepAliveWorker : Starting.")
for true { for true {