eleksir/buny-jabber-bot
0
Fork 0
Code Pull Requests Activity

Do not os.Exit() on "wire" and xmpp-proto errors, re-run main routine and helper goroutines

Browse Source
This commit is contained in:
eleksir
2023-08-05 20:02:10 -07:00
parent e8f8b3cb49
commit 267dd841a4
11 changed files with 606 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
buny.go
View File

@@ -1,7 +1,8 @@
package main package main
import ( import (
"os" "errors"
"fmt"
"regexp" "regexp"
"strings" "strings"
@@ -84,8 +85,17 @@ func buny(v xmpp.Presence) error { //nolint:gocognit
"http://jabber.org/protocol/muc#admin", "http://jabber.org/protocol/muc#admin",
ban, ban,
); err != nil { ); err != nil {
log.Errorf("Unable to ban user: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to ban user: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
continue
} }
return err return err
@@ -121,8 +131,17 @@ func buny(v xmpp.Presence) error { //nolint:gocognit
"http://jabber.org/protocol/muc#admin", "http://jabber.org/protocol/muc#admin",
ban, ban,
); err != nil { ); err != nil {
log.Errorf("Unable to ban user: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to ban user: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
continue
} }
return err return err

134
event_parser.go
View File

@@ -2,8 +2,8 @@ package main
import ( import (
"encoding/xml" "encoding/xml"
"errors"
"fmt" "fmt"
"os"
"strings" "strings"
"time" "time"
@@ -111,8 +111,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send version info to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send version info to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -128,8 +137,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
"", "",
); err != nil { ); err != nil {
log.Errorf("Unable to send pong to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send pong to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -151,8 +169,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send last activity time to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send last activity time to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -176,8 +203,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send disco#info to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send disco#info to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -199,8 +235,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeError, xmpp.IQTypeError,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send disco#items to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send disco#items to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -237,8 +282,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send urn:xmpp:time to jabber server: id=%s, err=%s", id, err) err := errors.New(
os.Exit(1) fmt.Sprintf(
"Unable to send urn:xmpp:time to jabber server: id=%s, err=%s",
id,
err,
),
)
gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -275,13 +329,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeError, xmpp.IQTypeError,
answer, answer,
); err != nil { ); err != nil {
log.Errorf( err := errors.New(
"Unable to send pubsub feature unimplemented to jabber server: id=%s, err=%s", fmt.Sprintf(
id, "Unable to send pubsub feature unimplemented to jabber server: id=%s, err=%s",
err, id,
err,
),
) )
os.Exit(1) gTomb.Kill(err)
return
} }
parseError = false parseError = false
@@ -312,9 +370,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeResult, xmpp.IQTypeResult,
"", "",
); err != nil { ); err != nil {
log.Errorf("Unable to send pong to jabber server: id=%s, err=%s", id, err) err := errors.New(
fmt.Sprintf(
"Unable to send pong to jabber server: id=%s, err=%s",
id,
err,
),
)
os.Exit(1) gTomb.Kill(err)
return
} }
} }
@@ -394,12 +460,17 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
xmpp.IQTypeError, xmpp.IQTypeError,
answer, answer,
); err != nil { ); err != nil {
log.Errorf("Unable to send set feature unimplemented to jabber server: id=%s, err=%s", err := errors.New(
id, fmt.Sprintf(
err, "Unable to send set feature unimplemented to jabber server: id=%s, err=%s",
id,
err,
),
) )
os.Exit(1) gTomb.Kill(err)
return
} }
log.Debug(spew.Sdump(e)) log.Debug(spew.Sdump(e))
@@ -449,9 +520,16 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
time.Sleep(time.Duration(config.Jabber.MucRejoinDelay) * time.Second) time.Sleep(time.Duration(config.Jabber.MucRejoinDelay) * time.Second)
if _, err := talk.JoinMUCNoHistory(iqErrorCancelNotAcceptable.By, nick); err != nil { if _, err := talk.JoinMUCNoHistory(iqErrorCancelNotAcceptable.By, nick); err != nil {
log.Errorf("Looks like connection to server also lost") err := errors.New(
fmt.Sprintf(
"Looks like connection to server also lost err=%s",
err,
),
)
os.Exit(1) gTomb.Kill(err)
return
} }
} else { } else {
log.Error( log.Error(
@@ -519,7 +597,9 @@ func parseEvent(e interface{}) { //nolint:gocognit,gocyclo
} }
if err := buny(v); err != nil { if err := buny(v); err != nil {
os.Exit(1) gTomb.Kill(err)
return
} }
} }

10
globals.go
View File

@@ -4,6 +4,7 @@ import (
"os" "os"
"github.com/eleksir/go-xmpp" "github.com/eleksir/go-xmpp"
"gopkg.in/tomb.v1"
) )
// Config - это у нас глобальная штука :) . // Config - это у нас глобальная штука :) .
@@ -45,16 +46,16 @@ var lastServerActivity int64
// Время последней активности MUC-ов, нужно для пингов - посылаем пинги, только если давненько ничего не приходило из // Время последней активности MUC-ов, нужно для пингов - посылаем пинги, только если давненько ничего не приходило из
// muc-ов. // muc-ов.
var lastMucActivity = NewCollection() var lastMucActivity *Collection
// Получен ли ответ на запрос disco#info к серверу. // Получен ли ответ на запрос disco#info к серверу.
var serverCapsQueried bool var serverCapsQueried bool
// sync.Map-ка с капабилити сервера. // sync.Map-ка с капабилити сервера.
var serverCapsList = NewCollection() var serverCapsList *Collection
// sync.Map-ка с комнатами и их capability. // sync.Map-ка с комнатами и их capability.
var mucCapsList = NewCollection() var mucCapsList *Collection
// Время, когда был отправлен c2s ping. // Время, когда был отправлен c2s ping.
var serverPingTimestampTx int64 var serverPingTimestampTx int64
@@ -62,4 +63,7 @@ var serverPingTimestampTx int64
// Время, когда был принят s2c pong. // Время, когда был принят s2c pong.
var serverPingTimestampRx int64 var serverPingTimestampRx int64
// Объектик для хранения стейта утилизатора горутинок
var gTomb tomb.Tomb
/* vim: set ft=go noet ai ts=4 sw=4 sts=4: */ /* vim: set ft=go noet ai ts=4 sw=4 sts=4: */

1
go.mod
View File

@@ -8,6 +8,7 @@ require (
github.com/hjson/hjson-go v3.3.0+incompatible github.com/hjson/hjson-go v3.3.0+incompatible
github.com/sirupsen/logrus v1.9.3 github.com/sirupsen/logrus v1.9.3
golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691 golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7
) )
require ( require (

2
go.sum
View File

@@ -20,5 +20,7 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA= golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

272
lib.go
View File

@@ -2,6 +2,7 @@ package main
import ( import (
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"math/rand" "math/rand"
"os" "os"
@@ -452,20 +453,15 @@ func establishConnection() {
talk, err = options.NewClient() talk, err = options.NewClient()
if err != nil { if err != nil {
log.Error(err) gTomb.Kill(err)
return
os.Exit(1)
} }
// По идее keepalive должен же проходить только, если мы уже на сервере, так? // По идее keepalive должен же проходить только, если мы уже на сервере, так?
if _, err := talk.SendKeepAlive(); err != nil { if _, err := talk.SendKeepAlive(); err != nil {
log.Errorf("Try to send initial KeepAlive, got error: %s", err) log.Errorf("Try to send initial KeepAlive, got error: %s", err)
time.Sleep(time.Duration(config.Jabber.ReconnectDelay) * time.Second)
// Понятно, что мы уже не в процессе установления соединения, а произошла ошибка gTomb.Kill(err)
connecting = false
os.Exit(1)
} }
log.Info("Connected") log.Info("Connected")
@@ -481,8 +477,9 @@ func establishConnection() {
}, },
); err != nil { ); err != nil {
log.Infof("Unable to send presence to jabber server: %s", err) log.Infof("Unable to send presence to jabber server: %s", err)
gTomb.Kill(err)
os.Exit(1) return
} }
lastActivity = time.Now().Unix() lastActivity = time.Now().Unix()
@@ -495,8 +492,9 @@ func establishConnection() {
if err != nil { if err != nil {
log.Infof("Unable to send disco#info to jabber server: %s", err) log.Infof("Unable to send disco#info to jabber server: %s", err)
gTomb.Kill(err)
os.Exit(1) return
} }
} }
@@ -507,7 +505,7 @@ func joinMuc(room string) {
if _, err := talk.DiscoverInfo(talk.JID(), room); err != nil { if _, err := talk.DiscoverInfo(talk.JID(), room); err != nil {
log.Infof("Unable to send disco#info to MUC %s: %s", room, err) log.Infof("Unable to send disco#info to MUC %s: %s", room, err)
os.Exit(1) gTomb.Kill(err)
} }
// Ждём, пока muc нам вернёт список фичей. // Ждём, пока muc нам вернёт список фичей.
@@ -539,7 +537,7 @@ func joinMuc(room string) {
if _, err := talk.JoinMUCNoHistory(room, config.Jabber.Nick); err != nil { if _, err := talk.JoinMUCNoHistory(room, config.Jabber.Nick); err != nil {
log.Errorf("Unable to join to MUC: %s", room) log.Errorf("Unable to join to MUC: %s", room)
os.Exit(1) gTomb.Kill(err)
} }
log.Infof("Joining to MUC: %s", room) log.Infof("Joining to MUC: %s", room)
@@ -577,95 +575,114 @@ func joinMuc(room string) {
}, },
); err != nil { ); err != nil {
log.Infof("Unable to send presence to MUC %s: %s", room, err) log.Infof("Unable to send presence to MUC %s: %s", room, err)
os.Exit(1) gTomb.Kill(err)
return
} }
} }
// probeServerLiveness() проверяет живость соединения с сервером. Для многих серверов обязательная штука, без которой // probeServerLiveness() проверяет живость соединения с сервером. Для многих серверов обязательная штука, без которой
// они выкидывают клиента через некоторое время неактивности. // они выкидывают клиента через некоторое время неактивности.
func probeServerLiveness() { //nolint:gocognit func probeServerLiveness() { //nolint:gocognit
defer gTomb.Done()
for { for {
if shutdown { select {
case <-gTomb.Dying():
return return
}
sleepTime := time.Duration(config.Jabber.ServerPingDelay) * 1000 * time.Millisecond default:
sleepTime += time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec for {
time.Sleep(sleepTime) if shutdown {
return
}
if !isConnected { sleepTime := time.Duration(config.Jabber.ServerPingDelay) * 1000 * time.Millisecond
continue sleepTime += time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec
} time.Sleep(sleepTime)
// Пингуем, только если не было никакой активности в течение > config.Jabber.ServerPingDelay, if !isConnected {
// в худшем случе это будет ~ (config.Jabber.PingSplayDelay * 2) + config.Jabber.PingSplayDelay continue
// if (time.Now().Unix() - lastServerActivity) < (config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay) { }
// continue
// }
if serverCapsQueried { // Сервер ответил на disco#info // Пингуем, только если не было никакой активности в течение > config.Jabber.ServerPingDelay,
var ( // в худшем случе это будет ~ (config.Jabber.PingSplayDelay * 2) + config.Jabber.PingSplayDelay
value interface{} // if (time.Now().Unix() - lastServerActivity) < (config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay) {
exist bool // continue
) // }
value, exist = serverCapsList.Get("urn:xmpp:ping") if serverCapsQueried { // Сервер ответил на disco#info
var (
value interface{}
exist bool
)
switch { value, exist = serverCapsList.Get("urn:xmpp:ping")
// Сервер анонсировал, что умеет в c2s пинги
case exist && value.(bool):
// Таймаут c2s пинга. Возьмём сумму задержки между пингами, добавим таймаут коннекта и добавим
// максимальную корректировку разброса.
txTimeout := config.Jabber.ServerPingDelay + config.Jabber.ConnectionTimeout
txTimeout += config.Jabber.PingSplayDelay
rxTimeout := txTimeout
rxTimeAgo := time.Now().Unix() - serverPingTimestampRx
if serverPingTimestampTx > 0 { // Первая пуля от нас ушла...
switch { switch {
// Давненько мы не получали понгов от сервера, вероятно, соединение с сервером утеряно? // Сервер анонсировал, что умеет в c2s пинги
case rxTimeAgo > (rxTimeout * 2): case exist && value.(bool):
log.Debugf( // Таймаут c2s пинга. Возьмём сумму задержки между пингами, добавим таймаут коннекта и добавим
"Stall connection detected. No c2s pong for %d seconds", // максимальную корректировку разброса.
rxTimeAgo, txTimeout := config.Jabber.ServerPingDelay + config.Jabber.ConnectionTimeout
) txTimeout += config.Jabber.PingSplayDelay
rxTimeout := txTimeout
os.Exit(1) rxTimeAgo := time.Now().Unix() - serverPingTimestampRx
// По-умолчанию, мы отправляем c2s пинг if serverPingTimestampTx > 0 { // Первая пуля от нас ушла...
default: switch {
log.Debugf("Sending c2s ping from %s to %s", talk.JID(), config.Jabber.Server) // Давненько мы не получали понгов от сервера, вероятно, соединение с сервером утеряно?
case rxTimeAgo > (rxTimeout * 2):
err := errors.New(
fmt.Sprintf(
"Stall connection detected. No c2s pong for %d seconds",
rxTimeAgo,
),
)
if err := talk.PingC2S(talk.JID(), config.Jabber.Server); err != nil { gTomb.Kill(err)
os.Exit(1) continue
// По-умолчанию, мы отправляем c2s пинг
default:
log.Debugf("Sending c2s ping from %s to %s", talk.JID(), config.Jabber.Server)
if err := talk.PingC2S(talk.JID(), config.Jabber.Server); err != nil {
gTomb.Kill(err)
continue
}
serverPingTimestampTx = time.Now().Unix()
}
} else { // Первая пуля пока не вылетела, отправляем
log.Debugf("Sending first c2s ping from %s to %s", talk.JID(), config.Jabber.Server)
if err := talk.PingC2S(talk.JID(), config.Jabber.Server); err != nil {
gTomb.Kill(err)
continue
}
serverPingTimestampTx = time.Now().Unix()
} }
serverPingTimestampTx = time.Now().Unix() // Сервер не анонсировал, что умеет в c2s пинги
default:
log.Debug("Sending keepalive whitespace ping")
if _, err := talk.SendKeepAlive(); err != nil {
gTomb.Kill(err)
continue
}
} }
} else { // Первая пуля пока не вылетела, отправляем } else { // Сервер не ответил на disco#info
log.Debugf("Sending first c2s ping from %s to %s", talk.JID(), config.Jabber.Server) log.Debug("Sending keepalive whitespace ping")
if err := talk.PingC2S(talk.JID(), config.Jabber.Server); err != nil { if _, err := talk.SendKeepAlive(); err != nil {
os.Exit(1) gTomb.Kill(err)
continue
} }
serverPingTimestampTx = time.Now().Unix()
} }
// Сервер не анонсировал, что умеет в c2s пинги
default:
log.Debug("Sending keepalive whitespace ping")
if _, err := talk.SendKeepAlive(); err != nil {
os.Exit(1)
}
}
} else { // Сервер не ответил на disco#info
log.Debug("Sending keepalive whitespace ping")
if _, err := talk.SendKeepAlive(); err != nil {
os.Exit(1)
} }
} }
} }
@@ -673,58 +690,69 @@ func probeServerLiveness() { //nolint:gocognit
// probeMUCLiveness() Пингует MUC-и, нужно для проверки, что клиент ещё находится в MUC-е. // probeMUCLiveness() Пингует MUC-и, нужно для проверки, что клиент ещё находится в MUC-е.
func probeMUCLiveness() { func probeMUCLiveness() {
defer gTomb.Done()
for { for {
for _, room := range roomsConnected { select {
var ( case <-gTomb.Dying():
exist bool return
lastActivityTS interface{}
)
// Если записи про комнату нету, то пинговать её бессмысленно. default:
if lastActivityTS, exist = lastMucActivity.Get(room); !exist { for {
continue for _, room := range roomsConnected {
} var (
exist bool
lastActivityTS interface{}
)
// Если время последней активности в чятике не превысило // Если записи про комнату нету, то пинговать её бессмысленно.
// config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay, ничего не пингуем. if lastActivityTS, exist = lastMucActivity.Get(room); !exist {
if (time.Now().Unix() - lastActivityTS.(int64)) < (config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay) { continue
continue
}
/* Пинг MUC-а по сценарию без серверной оптимизации мы реализовывать не будем. Это как-то не надёжно.
go func(room string) {
// Небольшая рандомная задержка перед пингом комнаты
sleepTime := time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec
time.Sleep(sleepTime)
if err := talk.PingS2S(talk.JID(), room+"/"+config.Jabber.Nick); err != nil {
reEstablishConnection()
}
}(room)
*/
var roomMap interface{}
roomMap, exist = mucCapsList.Get(room)
// Пинги комнаты проводим, только если она записана, как прошедшая disco#info и поддерживающая
// Server Optimization.
if exist && roomMap.(map[string]bool)["http://jabber.org/protocol/muc#self-ping-optimization"] {
go func(room string) {
// Небольшая рандомная задержка перед пингом комнаты.
sleepTime := time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec
time.Sleep(sleepTime)
log.Debugf("Sending MUC ping from %s to %s", talk.JID(), room)
if err := talk.PingS2S(talk.JID(), room); err != nil {
os.Exit(1)
} }
}(room)
// Если время последней активности в чятике не превысило
// config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay, ничего не пингуем.
if (time.Now().Unix() - lastActivityTS.(int64)) < (config.Jabber.ServerPingDelay + config.Jabber.PingSplayDelay) {
continue
}
/* Пинг MUC-а по сценарию без серверной оптимизации мы реализовывать не будем. Это как-то не надёжно.
go func(room string) {
// Небольшая рандомная задержка перед пингом комнаты
sleepTime := time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec
time.Sleep(sleepTime)
if err := talk.PingS2S(talk.JID(), room+"/"+config.Jabber.Nick); err != nil {
gTomb.Kill(err)
continue
}
}(room)
*/
var roomMap interface{}
roomMap, exist = mucCapsList.Get(room)
// Пинги комнаты проводим, только если она записана, как прошедшая disco#info и поддерживающая
// Server Optimization.
if exist && roomMap.(map[string]bool)["http://jabber.org/protocol/muc#self-ping-optimization"] {
go func(room string) {
// Небольшая рандомная задержка перед пингом комнаты.
sleepTime := time.Duration(rand.Int63n(1000*config.Jabber.PingSplayDelay)) * time.Millisecond //nolint:gosec
time.Sleep(sleepTime)
log.Debugf("Sending MUC ping from %s to %s", talk.JID(), room)
if err := talk.PingS2S(talk.JID(), room); err != nil {
gTomb.Kill(err)
}
}(room)
}
}
time.Sleep(time.Duration(config.Jabber.MucPingDelay) * time.Second)
} }
} }
time.Sleep(time.Duration(config.Jabber.MucPingDelay) * time.Second)
} }
} }

166
main.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/eleksir/go-xmpp" "github.com/eleksir/go-xmpp"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"gopkg.in/tomb.v1"
) )
func init() { func init() {
@@ -43,12 +44,6 @@ func init() {
log.SetLevel(log.InfoLevel) log.SetLevel(log.InfoLevel)
} }
// Пока про capabilities мы ничего не знаем
serverCapsQueried = false
// Мы пока что не пинговали сервер.
serverPingTimestampRx = 0
serverPingTimestampTx = 0
} }
func main() { func main() {
@@ -79,77 +74,122 @@ func main() {
// github.com/mattn/go-xmpp пишет в stdio, нам этого не надо, ловим выхлоп его в logrus с уровнем trace // github.com/mattn/go-xmpp пишет в stdio, нам этого не надо, ловим выхлоп его в logrus с уровнем trace
xmpp.DebugWriter = log.WithFields(log.Fields{"logger": "stdlib"}).WriterLevel(log.TraceLevel) xmpp.DebugWriter = log.WithFields(log.Fields{"logger": "stdlib"}).WriterLevel(log.TraceLevel)
options = &xmpp.Options{ //nolint:exhaustruct // Хэндлер сигналов не надо трогать, он нужен для завершения программы целиком
Host: fmt.Sprintf("%s:%d", config.Jabber.Server, config.Jabber.Port),
User: config.Jabber.User,
Password: config.Jabber.Password,
Resource: config.Jabber.Resource,
NoTLS: !config.Jabber.Ssl,
StartTLS: config.Jabber.StartTLS,
TLSConfig: &tls.Config{ //nolint:exhaustruct
ServerName: config.Jabber.Server,
InsecureSkipVerify: !config.Jabber.SslVerify, //nolint:gosec
},
InsecureAllowUnencryptedAuth: config.Jabber.InsecureAllowUnencryptedAuth,
Debug: verboseClient,
Session: false,
Status: "xa",
StatusMessage: fmt.Sprintf("%s bot is starting up", config.Jabber.Nick),
DialTimeout: time.Duration(config.Jabber.ConnectionTimeout) * time.Second,
}
go sigHandler() go sigHandler()
signal.Notify(sigChan, os.Interrupt) signal.Notify(sigChan, os.Interrupt)
// Установим коннект
establishConnection()
serverPingTimestampRx = time.Now().Unix() // Считаем, что если коннект запустился, то первый пинг успешен
// Тыкаем сервер палочкой, проверяем, что коннект жив и переустанавливаем его, если он не жив
go probeServerLiveness()
// Тыкаем muc-и палочкой, проверяем, что они живы и пере-заходим в них, если пинги пропали
go probeMUCLiveness()
// Гребём сообщения...
for { for {
// Стриггерилось завершение работы приложения, или соединение не установлено (порвалось, например) options = &xmpp.Options{ //nolint:exhaustruct
// грести не надо Host: fmt.Sprintf("%s:%d", config.Jabber.Server, config.Jabber.Port),
if shutdown { User: config.Jabber.User,
break Password: config.Jabber.Password,
Resource: config.Jabber.Resource,
NoTLS: !config.Jabber.Ssl,
StartTLS: config.Jabber.StartTLS,
TLSConfig: &tls.Config{ //nolint:exhaustruct
ServerName: config.Jabber.Server,
InsecureSkipVerify: !config.Jabber.SslVerify, //nolint:gosec
},
InsecureAllowUnencryptedAuth: config.Jabber.InsecureAllowUnencryptedAuth,
Debug: verboseClient,
Session: false,
Status: "xa",
StatusMessage: fmt.Sprintf("%s bot is starting up", config.Jabber.Nick),
DialTimeout: time.Duration(config.Jabber.ConnectionTimeout) * time.Second,
} }
if !isConnected { // Через tomb попробуем попробуем сделать выход горутинок управляемым
continue gTomb = tomb.Tomb{}
}
chat, err := talk.Recv() // Устанавливаем соединение и гребём события, посылаемые сервером
myLoop()
if err != nil { log.Error(gTomb.Wait())
log.Errorf("Unable to get events from server: %s", err) time.Sleep(time.Duration(config.Jabber.ReconnectDelay) * time.Second)
}
}
switch { func myLoop() {
// Стрим не читается, он закрылся с той стороны во время чтения defer gTomb.Done()
case errors.Is(err, io.EOF):
os.Exit(1)
// Пытаемся читать закрытый сокет for {
case errors.Is(err, net.ErrClosed): select {
os.Exit(1) case <-gTomb.Dying():
return
default:
// Зададим начальное значение глобальным переменным
serverPingTimestampRx = 0
serverPingTimestampTx = 0
roomsConnected = make([]string, 1)
lastActivity = 0
lastServerActivity = 0
lastMucActivity = NewCollection()
serverCapsQueried = false
serverCapsList = NewCollection()
mucCapsList = NewCollection()
serverPingTimestampTx = 0
serverPingTimestampRx = 0
// Не смогли записать в сокет // Установим коннект
case errors.Is(err, net.ErrWriteToConnected): establishConnection()
os.Exit(1)
// Не сетевая проблема serverPingTimestampRx = time.Now().Unix() // Считаем, что если коннект запустился, то первый пинг успешен
default:
// Это уже что-то странное. // Тыкаем сервер палочкой, проверяем, что коннект жив и переустанавливаем его, если он не жив
// Вероятно, ошибка парсинга xml. Собственно, баг сервера, тут мы ничего поделать не можем go probeServerLiveness()
os.Exit(1)
// Тыкаем muc-и палочкой, проверяем, что они живы и пере-заходим в них, если пинги пропали
go probeMUCLiveness()
// Гребём сообщения...
for {
// Стриггерилось завершение работы приложения, или соединение не установлено (порвалось, например)
// грести не надо
if shutdown {
break
}
if !isConnected {
continue
}
chat, err := talk.Recv()
if err != nil {
log.Errorf("Unable to get events from server: %s", err)
switch {
// Стрим не читается, он закрылся с той стороны во время чтения
case errors.Is(err, io.EOF):
err := errors.New("Tcp stream closed while reading")
gTomb.Kill(err)
continue
// Пытаемся читать закрытый сокет
case errors.Is(err, net.ErrClosed):
err := errors.New("Unable to read closed socket")
gTomb.Kill(err)
continue
// Не смогли записать в сокет
case errors.Is(err, net.ErrWriteToConnected):
err := errors.New("Unable to write to socket")
gTomb.Kill(err)
continue
// Не сетевая проблема
default:
// Это уже что-то странное.
// Вероятно, ошибка парсинга xml. Собственно, баг сервера, тут мы ничего поделать не можем
err := errors.New("error during parsing received message")
gTomb.Kill(err)
continue
}
}
parseEvent(chat)
} }
} }
parseEvent(chat)
} }
} }

29
vendor/gopkg.in/tomb.v1/LICENSE generated vendored Normal file
View File

@@ -0,0 +1,29 @@
tomb - support for clean goroutine termination in Go.
Copyright (c) 2010-2011 - Gustavo Niemeyer <gustavo@niemeyer.net>
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
* Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

4
vendor/gopkg.in/tomb.v1/README.md generated vendored Normal file
View File

@@ -0,0 +1,4 @@
Installation and usage
----------------------
See [gopkg.in/tomb.v1](https://gopkg.in/tomb.v1) for documentation and usage details.

176
vendor/gopkg.in/tomb.v1/tomb.go generated vendored Normal file
View File

@@ -0,0 +1,176 @@
// Copyright (c) 2011 - Gustavo Niemeyer <gustavo@niemeyer.net>
//
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
// * Redistributions of source code must retain the above copyright notice,
// this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
// * Neither the name of the copyright holder nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
// The tomb package offers a conventional API for clean goroutine termination.
//
// A Tomb tracks the lifecycle of a goroutine as alive, dying or dead,
// and the reason for its death.
//
// The zero value of a Tomb assumes that a goroutine is about to be
// created or already alive. Once Kill or Killf is called with an
// argument that informs the reason for death, the goroutine is in
// a dying state and is expected to terminate soon. Right before the
// goroutine function or method returns, Done must be called to inform
// that the goroutine is indeed dead and about to stop running.
//
// A Tomb exposes Dying and Dead channels. These channels are closed
// when the Tomb state changes in the respective way. They enable
// explicit blocking until the state changes, and also to selectively
// unblock select statements accordingly.
//
// When the tomb state changes to dying and there's still logic going
// on within the goroutine, nested functions and methods may choose to
// return ErrDying as their error value, as this error won't alter the
// tomb state if provided to the Kill method. This is a convenient way to
// follow standard Go practices in the context of a dying tomb.
//
// For background and a detailed example, see the following blog post:
//
// http://blog.labix.org/2011/10/09/death-of-goroutines-under-control
//
// For a more complex code snippet demonstrating the use of multiple
// goroutines with a single Tomb, see:
//
// http://play.golang.org/p/Xh7qWsDPZP
//
package tomb
import (
"errors"
"fmt"
"sync"
)
// A Tomb tracks the lifecycle of a goroutine as alive, dying or dead,
// and the reason for its death.
//
// See the package documentation for details.
type Tomb struct {
m sync.Mutex
dying chan struct{}
dead chan struct{}
reason error
}
var (
ErrStillAlive = errors.New("tomb: still alive")
ErrDying = errors.New("tomb: dying")
)
func (t *Tomb) init() {
t.m.Lock()
if t.dead == nil {
t.dead = make(chan struct{})
t.dying = make(chan struct{})
t.reason = ErrStillAlive
}
t.m.Unlock()
}
// Dead returns the channel that can be used to wait
// until t.Done has been called.
func (t *Tomb) Dead() <-chan struct{} {
t.init()
return t.dead
}
// Dying returns the channel that can be used to wait
// until t.Kill or t.Done has been called.
func (t *Tomb) Dying() <-chan struct{} {
t.init()
return t.dying
}
// Wait blocks until the goroutine is in a dead state and returns the
// reason for its death.
func (t *Tomb) Wait() error {
t.init()
<-t.dead
t.m.Lock()
reason := t.reason
t.m.Unlock()
return reason
}
// Done flags the goroutine as dead, and should be called a single time
// right before the goroutine function or method returns.
// If the goroutine was not already in a dying state before Done is
// called, it will be flagged as dying and dead at once with no
// error.
func (t *Tomb) Done() {
t.Kill(nil)
close(t.dead)
}
// Kill flags the goroutine as dying for the given reason.
// Kill may be called multiple times, but only the first
// non-nil error is recorded as the reason for termination.
//
// If reason is ErrDying, the previous reason isn't replaced
// even if it is nil. It's a runtime error to call Kill with
// ErrDying if t is not in a dying state.
func (t *Tomb) Kill(reason error) {
t.init()
t.m.Lock()
defer t.m.Unlock()
if reason == ErrDying {
if t.reason == ErrStillAlive {
panic("tomb: Kill with ErrDying while still alive")
}
return
}
if t.reason == nil || t.reason == ErrStillAlive {
t.reason = reason
}
// If the receive on t.dying succeeds, then
// it can only be because we have already closed it.
// If it blocks, then we know that it needs to be closed.
select {
case <-t.dying:
default:
close(t.dying)
}
}
// Killf works like Kill, but builds the reason providing the received
// arguments to fmt.Errorf. The generated error is also returned.
func (t *Tomb) Killf(f string, a ...interface{}) error {
err := fmt.Errorf(f, a...)
t.Kill(err)
return err
}
// Err returns the reason for the goroutine death provided via Kill
// or Killf, or ErrStillAlive when the goroutine is still alive.
func (t *Tomb) Err() (reason error) {
t.init()
t.m.Lock()
reason = t.reason
t.m.Unlock()
return
}

3
vendor/modules.txt vendored
View File

@@ -22,3 +22,6 @@ golang.org/x/exp/slices
golang.org/x/sys/internal/unsafeheader golang.org/x/sys/internal/unsafeheader
golang.org/x/sys/unix golang.org/x/sys/unix
golang.org/x/sys/windows golang.org/x/sys/windows
# gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7
## explicit
gopkg.in/tomb.v1