0

Есть функция, которая запущена в горутине:

func (c *controlUC) WebhookPool() {
    for {
        if len(c.webhookPool) == 0 {
            continue
        }
        for i := 0; i < len(c.webhookPool); i++ {
            if !c.webhookPool[i].LastSentTime.IsZero() && time.Now().Before(c.webhookPool[i].LastSentTime.Add(GetDelayBySentCount(c.webhookPool[i].SendCount))) {
                continue
            }
            var headers = make(map[string]string)
            headers["Content-type"] = "application/json"
            _, statusCode, err := c.fhttpClient.Request("POST", c.webhookPool[i].Path, c.webhookPool[i].Body, nil, headers)
            if err != nil {
                c.logger.Error(err)
                return
            }
            if statusCode != 200 {
                if c.webhookPool[i].SendCount >= 2 {
                    c.webhookPool = append(c.webhookPool[:i], c.webhookPool[i+1:]...)
                    i--
                    continue
                }
                c.webhookPool[i].SendCount++
            } else {
                c.webhookPool = append(c.webhookPool[:i], c.webhookPool[i+1:]...)
                i--
                continue
            }
            c.webhookPool[i].LastSentTime = time.Now()
        }
    }
}
// webhookPool []models.WebhookPoolElem
type WebhookPoolElem struct {
    SendCount    int
    LastSentTime time.Time
    Path         string
    Body         []byte
}

В c.webhookPool добавляется элемент webhookPoolElem, после чего на сервер (путь берется из WebhookPoolElem.path) отправляется запрос. Если сервер вернул не 200 код, тогда нужно еще раз отправить запрос, через X секунд (берется из GetDelayBySentCount (), в зависимости от SendCount возвращает разное время). Количество попыток ограничено (c.webhookPool[i].SendCount >= 2).

Но, может, эту функцию нужно через каналы делать? Если да, то как?

1 ответ 1

1

надеюсь, я правильно понял вашу мысль



import (
    "testing"
    "time"

    "github.com/valyala/fasthttp"
)

type controlUC struct {
    webhookPool chan WebhookPoolElem
    logger      func(format string, args ...interface{})
    fhttpClient *fasthttp.Client
}

func newControlUC(webhookPool chan WebhookPoolElem, log func(format string, args ...interface{})) *controlUC {
    return &controlUC{
        webhookPool: webhookPool,
        logger:      log,
        fhttpClient: &fasthttp.Client{},
    }
}

func GetDelayBySentCount(i int) time.Duration {
    return time.Duration(i) * time.Second
}

func (c *controlUC) WebhookPool() {
    for {
        // читаем из канала новый объект
        webhookPool, ok  := <-c.webhookPool

        if !ok {
            break
        }

        c.logger("[INFO] [%s] count is %v", webhookPool.Path, webhookPool.SendCount)

        // сразу же проверяем на условие
        if webhookPool.SendCount >= 2 {
            c.logger("[INFO] [%s] limit reached ", webhookPool.Path)
            continue // пропустить все, если условие выполнено
        }

        // дальше работайте как привыкли
        if !webhookPool.LastSentTime.IsZero() &&
            time.Now().Before(webhookPool.LastSentTime.Add(GetDelayBySentCount(webhookPool.SendCount))) {
            continue
        }
        var headers = make(map[string]string)
        headers["Content-type"] = "application/json"
        statusCode, _, err := c.fhttpClient.Get(nil, webhookPool.Path)
        if err != nil {
            c.logger("[ERROR] %v", err)
            return
        }

        // увеличит счетчик только если удалось отправить
        if statusCode == 200 {
            webhookPool.SendCount++
        }

        webhookPool.LastSentTime = time.Now()

        // снова добавляем в очередь для дальнейшей проверки
        go func() {
            c.webhookPool <- webhookPool
        }()
    }
}

// webhookPool []models.WebhookPoolElem
type WebhookPoolElem struct {
    SendCount    int
    LastSentTime time.Time
    Path         string
    Body         []byte
}

func TestIt(t *testing.T) {
    elems := []WebhookPoolElem{
        {
            SendCount: 0,
            Path:      "http://127.0.0.1:4040",
        },
        {
            SendCount: 0,
            Path:      "http://127.0.0.1:4040",
        },
        {
            SendCount: 0,
            Path:      "http://127.0.0.1:4040",
        },
    }

    webhookPoolCh :=  make(chan WebhookPoolElem, 0)

    // в отдельной горутине добавлять в канал данные
    go func(e []WebhookPoolElem , webhookPoolCh chan WebhookPoolElem) {
        for _, elem := range e {
            webhookPoolCh <- elem
        }
    }(elems, webhookPoolCh)

    uc := newControlUC(webhookPoolCh, t.Logf)

    uc.WebhookPool()
}

4
  • Похоже, что не правильно. Модифицировал немного - pastebin.com/bPRi6d5V (не работает, count всегда или 0 или 1). Канал создал буферизованным. Если удалось отправить, то сообщение больше не нужно оправлять, его нужно удалить из пула. А если if !elem.LastSentTime... тогда подождать, но не удалять сообщение из пула. 4 окт 2021 в 20:08
  • это мелочи. а пример с каналами вам ясен? 5 окт 2021 в 6:30
  • Да, спасибо. Понял почему пример не работал. Проблема была в том, что не заметил, что у меня функция каждые пять секунд добавляет по пять сообщений в пул, а я ожидал, что она один раз отработает. Финальный результат - pastebin.com/7HneQePP. Но работает только, если канал буферизованный. В этом случае нормально создавать буфер 5000? 5 окт 2021 в 8:22
  • да, нормально. 5000 не так уж и много. но смотрите, чтобы потребление памяти не росло 5 окт 2021 в 8:55

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими пользовательским соглашением, политикой конфиденциальности и политикой о куки

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.