0

Do not write data to dead socket

This commit is contained in:
eleksir
2024-04-19 21:29:20 +03:00
parent b0c784aa20
commit 3b6dd3fdef
4 changed files with 56 additions and 53 deletions

View File

@@ -54,7 +54,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
lastMucActivity.Set(muc, lastServerActivity) lastMucActivity.Set(muc, lastServerActivity)
} }
// Приватный чятик // Приватный чятик.
case "chat": case "chat":
// Здесь у нас может быть 2 вида From: // Здесь у нас может быть 2 вида From:
// slackware-current@conference.jabber.ru/eleksir - это если сообщение из публичного чятика // slackware-current@conference.jabber.ru/eleksir - это если сообщение из публичного чятика
@@ -70,7 +70,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
lastActivity = lastServerActivity lastActivity = lastServerActivity
// Внезапно, ошибки. По идее они должны ассоциироваться с отправляемыми сообщениями, но по ходу это // Внезапно, ошибки. По идее они должны ассоциироваться с отправляемыми сообщениями, но по ходу это
// не реализовано, поэтому мы получаем ошибки в форме отдельного чятика // не реализовано, поэтому мы получаем ошибки в форме отдельного чятика.
case "error": case "error":
// Это ошибка со стороны сервера, видимо, в ответ на наши действия и поскольку такие вещи происходят // Это ошибка со стороны сервера, видимо, в ответ на наши действия и поскольку такие вещи происходят
// асинхронно, связать их с конкретными нашими действиями довольно сложно. // асинхронно, связать их с конкретными нашими действиями довольно сложно.
@@ -80,7 +80,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
// IQ-сообщение - пинг, понг, что-то ещё... // IQ-сообщение - пинг, понг, что-то ещё...
case xmpp.IQ: case xmpp.IQ:
// По правилам IQ обязательно должна содержать ID, причём не пустой // По правилам IQ обязательно должна содержать ID, причём не пустой.
if v.ID == "" { if v.ID == "" {
log.Info("Got an IQ stanza with empty id, discarding") log.Info("Got an IQ stanza with empty id, discarding")
@@ -363,9 +363,9 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
} }
} }
// Поскольку никакой логики у нас на этот счёт не предусмотрено, то просто пропускаем ответ.
if mucNickMatch { if mucNickMatch {
log.Debugf("Got 2-nd stage MUC pong answer (xep-0410) from %s to %s", v.From, v.To) log.Debugf("Got 2-nd stage MUC pong answer (xep-0410) from %s to %s", v.From, v.To)
// Поскольку никакой логики у нас на этот счёт не предусмотрено, то просто пропускаем ответ
} else { } else {
log.Infof("Got unknown pong from %s to %s", v.From, v.To) log.Infof("Got unknown pong from %s to %s", v.From, v.To)
log.Debug(spew.Sdump(e)) log.Debug(spew.Sdump(e))
@@ -376,7 +376,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
log.Debug(spew.Sdump(e)) log.Debug(spew.Sdump(e))
} }
// Этот бот не управляется со стороны сервера, поэтому все попытки порулить игнорируем // Этот бот не управляется со стороны сервера, поэтому все попытки порулить игнорируем.
case xmpp.IQTypeSet: case xmpp.IQTypeSet:
if muc, _ := strings.CutSuffix(v.To, "/"); muc != "" { if muc, _ := strings.CutSuffix(v.To, "/"); muc != "" {
if slices.Contains(roomsConnected, muc) { if slices.Contains(roomsConnected, muc) {
@@ -404,7 +404,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
log.Debug(spew.Sdump(e)) log.Debug(spew.Sdump(e))
// Нам прилетело сообщение об ошибке // Нам прилетело сообщение об ошибке.
case xmpp.IQTypeError: case xmpp.IQTypeError:
// Если сервер не хочет пинговаться и отвечает ошибкой на пинг, то наверно он не умеет в пинги, // Если сервер не хочет пинговаться и отвечает ошибкой на пинг, то наверно он не умеет в пинги,
// хотя если мы его пингуем, значит он анонсировал такой capability. Вот, засранец! // хотя если мы его пингуем, значит он анонсировал такой capability. Вот, засранец!
@@ -432,7 +432,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
parseError = false parseError = false
} }
// Это у нас пинг xep-0410 и мы не в комнате, предполагается, что надо бы заджойниться // Это у нас пинг xep-0410 и мы не в комнате, предполагается, что надо бы заджойниться.
if parseError { if parseError {
var iqErrorCancelNotAcceptable jabberIqErrorCancelNotAcceptable var iqErrorCancelNotAcceptable jabberIqErrorCancelNotAcceptable
@@ -484,7 +484,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
log.Error("Unhandled IQ Error message") log.Error("Unhandled IQ Error message")
log.Error(spew.Sdump(e)) log.Error(spew.Sdump(e))
} }
// Нам прилетело что-то неизвестное из семейства IQ stanza // Нам прилетело что-то неизвестное из семейства IQ stanza.
default: default:
log.Info("Got an unknown IQ request. Dunno how deal with it, discarding") log.Info("Got an unknown IQ request. Dunno how deal with it, discarding")
log.Info(spew.Sdump(e)) log.Info(spew.Sdump(e))
@@ -505,7 +505,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
// клиента забанят из-за спама "вошёл-вышел". // клиента забанят из-за спама "вошёл-вышел".
// Однако, это не значит, что такую ситуацию мы не должны корректным образом обрабатывать. // Однако, это не значит, что такую ситуацию мы не должны корректным образом обрабатывать.
if v.Type == "unavailable" { if v.Type == "unavailable" {
// Считаем, что мы больше не в комнате, поэтому не знаем, кто там есть // Считаем, что мы больше не в комнате, поэтому не знаем, кто там есть.
roomPresences.Delete(v.From) roomPresences.Delete(v.From)
log.Error("Presence notification - looks like another instance of client leaves room") log.Error("Presence notification - looks like another instance of client leaves room")
@@ -548,6 +548,7 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
case "none": case "none":
if presenceJSONInterface, present := roomPresences.Get(room); present { if presenceJSONInterface, present := roomPresences.Get(room); present {
presenceJSONStrings := interfaceToStringSlice(presenceJSONInterface) presenceJSONStrings := interfaceToStringSlice(presenceJSONInterface)
var newPresenceJSONStrings []string var newPresenceJSONStrings []string
for _, presenceJSONstring := range presenceJSONStrings { for _, presenceJSONstring := range presenceJSONStrings {
@@ -565,9 +566,11 @@ func parseEvent(e interface{}) { //nolint:maintidx,gocognit,gocyclo
} }
// Участник пришёл // Участник пришёл
default: default:
var presenceJSONStrings []string var (
var newPresenceJSONStrings []string presenceJSONStrings []string
var presenceJSONBytes []byte newPresenceJSONStrings []string
presenceJSONBytes []byte
)
if presenceJSONInterface, present := roomPresences.Get(room); present { if presenceJSONInterface, present := roomPresences.Get(room); present {
presenceJSONStrings = interfaceToStringSlice(presenceJSONInterface) presenceJSONStrings = interfaceToStringSlice(presenceJSONInterface)

68
main.go
View File

@@ -26,7 +26,7 @@ func init() {
TimestampFormat: "2006-01-02 15:04:05", TimestampFormat: "2006-01-02 15:04:05",
}) })
// Добавим динамически формируемый путь до конфига в список возможных положений конфига в системе // Добавим динамически формируемый путь до конфига в список возможных положений конфига в системе.
executablePath, err := os.Executable() executablePath, err := os.Executable()
if err != nil { if err != nil {
@@ -84,14 +84,14 @@ func main() {
verboseClient = true verboseClient = true
} }
// 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)
// Хэндлер сигналов не надо трогать, он нужен для завершения программы целиком // Хэндлер сигналов не надо трогать, он нужен для завершения программы целиком.
go sigHandler() go sigHandler()
signal.Notify(sigChan, os.Interrupt) signal.Notify(sigChan, os.Interrupt)
// Иницализируем redis-клиента // Иницализируем redis-клиента.
redisClient = redis.NewClient(&redis.Options{ //nolint:exhaustruct redisClient = redis.NewClient(&redis.Options{ //nolint:exhaustruct
Addr: fmt.Sprintf("%s:%d", config.Redis.Server, config.Redis.Port), Addr: fmt.Sprintf("%s:%d", config.Redis.Server, config.Redis.Port),
}) })
@@ -121,10 +121,10 @@ func main() {
DialTimeout: time.Duration(config.Jabber.ConnectionTimeout) * time.Second, DialTimeout: time.Duration(config.Jabber.ConnectionTimeout) * time.Second,
} }
// Через tomb попробуем сделать выход горутинок управляемым // Через tomb попробуем сделать выход горутинок управляемым.
gTomb = tomb.Tomb{} gTomb = tomb.Tomb{}
// Устанавливаем соединение и гребём события, посылаемые сервером // Устанавливаем соединение и гребём события, посылаемые сервером.
myLoop() myLoop()
log.Error(gTomb.Wait()) log.Error(gTomb.Wait())
@@ -132,6 +132,7 @@ func main() {
// Если у нас wire error, то вызов .Close() повлечёт за собой ошибку, но мы вынуждены звать .Close(), чтоб // Если у нас wire error, то вызов .Close() повлечёт за собой ошибку, но мы вынуждены звать .Close(), чтоб
// закрыть tls контекст и почистить всё что связанно с прерванным соединением. // закрыть tls контекст и почистить всё что связанно с прерванным соединением.
log.Infoln("Closing connection to jabber server") log.Infoln("Closing connection to jabber server")
_ = talk.Close() _ = talk.Close()
time.Sleep(time.Duration(config.Jabber.ReconnectDelay) * time.Second) time.Sleep(time.Duration(config.Jabber.ReconnectDelay) * time.Second)
@@ -143,10 +144,11 @@ func myLoop() {
select { select {
case <-gTomb.Dying(): case <-gTomb.Dying():
gTomb.Done() gTomb.Done()
return return
default: default:
// Зададим начальное значение глобальным переменным // Зададим начальное значение глобальным переменным.
serverPingTimestampRx = 0 serverPingTimestampRx = 0
serverPingTimestampTx = 0 serverPingTimestampTx = 0
roomsConnected = make([]string, 1) roomsConnected = make([]string, 1)
@@ -160,65 +162,63 @@ func myLoop() {
serverPingTimestampRx = 0 serverPingTimestampRx = 0
roomPresences = NewCollection() roomPresences = NewCollection()
// Установим коннект // Установим коннект.
establishConnection() establishConnection()
serverPingTimestampRx = time.Now().Unix() // Считаем, что если коннект запустился, то первый пинг успешен serverPingTimestampRx = time.Now().Unix() // Считаем, что если коннект запустился, то первый пинг успешен.
// Тыкаем сервер палочкой, проверяем, что коннект жив и переустанавливаем его, если он не жив // Тыкаем сервер палочкой, проверяем, что коннект жив и переустанавливаем его, если он не жив.
go probeServerLiveness() go probeServerLiveness()
// Тыкаем muc-и палочкой, проверяем, что они живы и пере-заходим в них, если пинги пропали // Тыкаем muc-и палочкой, проверяем, что они живы и пере-заходим в них, если пинги пропали.
go probeMUCLiveness() go probeMUCLiveness()
// Гребём сообщения... // Гребём сообщения...
for { for {
// Стриггерилось завершение работы приложения, или соединение не установлено (порвалось, например) // Стриггерилось завершение работы приложения, или соединение не установлено (порвалось, например).
// грести не надо // грести не надо.
if shutdown { if shutdown {
break break
} }
if !isConnected { if !isConnected {
// Tight loop - это наверно не очень хорошо, думаю, ничего страшного не будет, если мы поспим 100мс.
time.Sleep(100 * time.Millisecond)
continue continue
} }
// Вынимаем ивент из "провода".
chat, err := talk.Recv() chat, err := talk.Recv()
if err != nil { if err != nil {
log.Errorf("Unable to get events from server: %s", err) log.Errorf("Unable to get events from server: %s", err)
// Попробуем скрасить печальные будни generic-ошибок...
switch { switch {
// Стрим не читается, он закрылся с той стороны во время чтения // Стрим не читается, он закрылся с той стороны во время чтения.
case errors.Is(err, io.EOF): case errors.Is(err, io.EOF):
err := fmt.Errorf("tcp stream closed while reading, err=%w", err) err = fmt.Errorf("tcp stream closed while reading, err=%w", err)
gTomb.Kill(err)
continue // Пытаемся читать закрытый сокет.
// Пытаемся читать закрытый сокет
case errors.Is(err, net.ErrClosed): case errors.Is(err, net.ErrClosed):
err := fmt.Errorf("unable to read closed socket, err=%w", err) err = fmt.Errorf("unable to read closed socket, err=%w", err)
gTomb.Kill(err)
continue // Не смогли записать в сокет.
// Не смогли записать в сокет
case errors.Is(err, net.ErrWriteToConnected): case errors.Is(err, net.ErrWriteToConnected):
err := fmt.Errorf("unable to write to socket, err=%w", err) err = fmt.Errorf("unable to write to socket, err=%w", err)
gTomb.Kill(err)
continue // Не сетевая проблема.
// Не сетевая проблема
default: default:
// Это уже что-то странное. // Это уже что-то странное.
// Вероятно, ошибка парсинга xml. Собственно, баг сервера, тут мы ничего поделать не можем // Вероятно, ошибка парсинга xml. Собственно, баг сервера, тут мы ничего поделать не можем.
err := fmt.Errorf("error during parsing received message, err=%w", err) err = fmt.Errorf("error during parsing received message, err=%w", err)
gTomb.Kill(err)
continue
} }
gTomb.Kill(err)
// Не забываем выходить на ошибке :(
return
} }
parseEvent(chat) parseEvent(chat)

View File

@@ -189,15 +189,15 @@ func readConfig() error { //nolint:gocognit,gocyclo
for n, channel := range sampleConfig.Jabber.Channels { for n, channel := range sampleConfig.Jabber.Channels {
if channel.Name == "" { if channel.Name == "" {
return errors.New("no \"name\" entry in jabber channel config") return fmt.Errorf("no \"name\" entry in jabber channel config")
} }
if channel.Nick == "" { if channel.Nick == "" {
sampleConfig.Jabber.Channels[n].Nick = sampleConfig.Jabber.Nick sampleConfig.Jabber.Channels[n].Nick = sampleConfig.Jabber.Nick
} }
// channel.Password может быть пустым, тогда пароля нет // channel.Password может быть пустым, тогда пароля нет.
} } //nolint:wsl
// Если список фраз с которыми стартует бот пустой, вносим в него 1 запись с пустой строкой // Если список фраз с которыми стартует бот пустой, вносим в него 1 запись с пустой строкой
if len(sampleConfig.Jabber.StartupStatus) == 0 { if len(sampleConfig.Jabber.StartupStatus) == 0 {
@@ -235,13 +235,13 @@ func readConfig() error { //nolint:gocognit,gocyclo
config = sampleConfig config = sampleConfig
configLoaded = true configLoaded = true
log.Infof("Using %s as config file", location) //nolint:wsl log.Infof("Using %s as config file", location)
break break
} }
if !configLoaded { if !configLoaded {
return errors.New("config was not loaded!") //nolint:goerr113,revive,stylecheck return fmt.Errorf("config was not loaded!") //nolint:goerr113,revive,stylecheck
} }
return err //nolint:wrapcheck return err //nolint:wrapcheck

View File

@@ -14,7 +14,7 @@ import (
// StoreKV сохраняет в указанной бд ключ и значение. // StoreKV сохраняет в указанной бд ключ и значение.
func StoreKV(db *pebble.DB, key string, value string) error { func StoreKV(db *pebble.DB, key string, value string) error {
var ( var (
kArray = []byte(key) //nolint:revive kArray = []byte(key)
vArray = []byte(value) vArray = []byte(value)
) )
@@ -26,7 +26,7 @@ func StoreKV(db *pebble.DB, key string, value string) error {
// FetchV достаёт значение по ключу. // FetchV достаёт значение по ключу.
func FetchV(db *pebble.DB, key string) (string, error) { func FetchV(db *pebble.DB, key string) (string, error) {
var ( var (
kArray = []byte(key) //nolint:revive kArray = []byte(key)
vArray []byte vArray []byte
valueString string valueString string
) )