test

workers.go

@@ -16,41 +16,62 @@ import (
 func MQGetMsgWorker(id int, msgs chan<- ChatWarsMessage) {
 	//log.Printf("MQGetMsgWorker[" + strconv.Itoa(id) + "] : Starting.")
 	var x ChatWarsMessage
-	conn, err := amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + cfg.Rabbit.Queue)
+	for true {
-	failOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to connect to RabbitMQ")
+		conn, err := amqp.Dial("amqps://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + cfg.Rabbit.Queue)
-	defer conn.Close()
+		logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to connect to RabbitMQ")
+		if err != nil {
+			conn.Close()
+			time.Sleep(15 * time.Second)
+		} else {
+			ch, err := conn.Channel()
+			logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to open a channel")
+			if err != nil {
+				ch.Close()
+				time.Sleep(15 * time.Second)
+			} else {
+				q, err := ch.QueueDeclare(
+					"msg", // 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 {
+					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, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to register a consumer")
+					if err != nil {
+						ch.Close()
+						conn.Close()
+						time.Sleep(15 * time.Second)
+					} else {
+						for d := range m {
+							// log.Printf("MQGetMsgWorker["+strconv.Itoa(id)+"] : Received a message: %s", string(d.Body))
+							err = json.Unmarshal(d.Body, &x)
+							logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Can't unmarshal.\n"+string(d.Body))
+							if err == nil {
+								msgs <- x
+							}
+						}
+						ch.Close()
+						conn.Close()
+					}
+				}
 
-	ch, err := conn.Channel()
+			}
-	failOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to open a channel")
-	defer ch.Close()
-
-	q, err := ch.QueueDeclare(
-		"msg", // name
-		false, // durable
-		false, // delete when unused
-		false, // exclusive
-		false, // no-wait
-		nil,   // arguments
-	)
-	failOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to declare a queue")
-
-	m, err := ch.Consume(
-		q.Name, // queue
-		"",     // consumer
-		true,   // auto-ack
-		false,  // exclusive
-		false,  // no-local
-		false,  // no-wait
-		nil,    // args
-	)
-	failOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Failed to register a consumer")
-
-	for d := range m {
-		// log.Printf("MQGetMsgWorker["+strconv.Itoa(id)+"] : Received a message: %s", string(d.Body))
-		err = json.Unmarshal(d.Body, &x)
-		logOnError(err, "MQGetMsgWorker["+strconv.Itoa(id)+"] : Can't unmarshal.\n"+string(d.Body))
-		if err == nil {
-			msgs <- x
 		}
 	}
 
@@ -347,87 +368,109 @@ func MQTGCmdWorker(id int, cmds <-chan TGCommand) {
 
 func MQKeepAliveWorker() {
 	//log.Printf("MQKeepAliveWorker : Starting.")
-	conn, err := amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + cfg.Rabbit.Queue)
+	for true {
-	failOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
+		conn, err := amqp.Dial("amqps://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + cfg.Rabbit.Queue)
-	defer conn.Close()
+		logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
-
+		if err != nil {
-	ch, err := conn.Channel()
+			conn.Close()
-	failOnError(err, "MQKeepAliveWorker : Failed to open a channel")
+			time.Sleep(15 * time.Second)
-	defer ch.Close()
+		} else {
-
+			ch, err := conn.Channel()
-	q, err := ch.QueueDeclare(
+			logOnError(err, "MQKeepAliveWorker : Failed to open a channel")
-		"keepalive", // name
+			if err != nil {
-		false,       // durable
+				ch.Close()
-		false,       // delete when unused
+				conn.Close()
-		false,       // exclusive
+				time.Sleep(15 * time.Second)
-		false,       // no-wait
-		nil,         // arguments
-	)
-	failOnError(err, "MQKeepAliveWorker : Failed to declare a queue")
-
-	m, err := ch.Consume(
-		q.Name, // queue
-		"",     // consumer
-		true,   // auto-ack
-		false,  // exclusive
-		false,  // no-local
-		false,  // no-wait
-		nil,    // args
-	)
-	failOnError(err, "MQKeepAliveWorker : Failed to register a consumer")
-
-	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{}
+				q, err := ch.QueueDeclare(
-				clt.Connection, err = amqp.Dial("amqp://" + cfg.Rabbit.User + ":" + cfg.Rabbit.Password + "@" + cfg.Rabbit.Host + "/" + x.Queue)
+					"keepalive", // name
-				logOnError(err, "MQKeepAliveWorker : Failed to connect to RabbitMQ")
+					false,       // durable
-				clt.Channel, err = clt.Connection.Channel()
+					false,       // delete when unused
-				logOnError(err, "MQKeepAliveWorker : Failed to open a channel")
+					false,       // exclusive
-				clt.Queue, err = clt.Channel.QueueDeclare(
+					false,       // no-wait
-					"msg", // name
+					nil,         // arguments
-					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)
+				if err != nil {
-				clientsQueue[x.UserID64] = &clt
+					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{
+									c := TGCommand{
-					Type:       commandSendMsg,
+										Type:       commandSendMsg,
-					ToUserID64: x.UserID64,
+										ToUserID64: x.UserID64,
-					Text:       "Your client is connected.",
+										Text:       "Your client is connected.",
-				}
+									}
-				TGCmdQueue <- c
+									TGCmdQueue <- c
-				c = TGCommand{
+									c = TGCommand{
-					Type:       commandSendMsg,
+										Type:       commandSendMsg,
-					ToUserID64: cfg.Bot.Admin,
+										ToUserID64: cfg.Bot.Admin,
-					Text:       fmt.Sprintf("Client `%s` is connected.", x.Nickname),
+										Text:       fmt.Sprintf("Client `%s` is connected.", x.Nickname),
-				}
+									}
-				TGCmdQueue <- c
+									TGCmdQueue <- c
 
-				clientSendCWMsg(x.UserID64, `🏅Me`)
+									clientSendCWMsg(x.UserID64, `🏅Me`)
-				/*
+									/*
-					c = TGCommand{
+										c = TGCommand{
-						Type:         commandSendMsg,
+											Type:         commandSendMsg,
-						FromUserID64: x.UserID64,
+											FromUserID64: x.UserID64,
-						ToChatID64:   userID64ChtWrsBot,
+											ToChatID64:   userID64ChtWrsBot,
-						Text:         `/hero`,
+											Text:         `/hero`,
+										}
+										MQTGCmdQueue <- c
+									*/
+								}
+							}
+						}
+						ch.Close()
+						conn.Close()
 					}
-					MQTGCmdQueue <- c
+				}
-				*/
+
 			}
 		}
 	}