asyncspammer

asyncspammer — это мини-библиотека на Go для построения конвейеров (pipeline), где этапы обмениваются данными через каналы.
Проект демонстрирует принципы конкурентного программирования, работу с sync.WaitGroup, семафорами и потоковую обработку данных без промежуточного накопления.

---

Возможности

• Универсальная функция RunPipeline для организации последовательной конвейерной обработки.
• Безопасное закрытие каналов без дедлоков.
• Контролируемый уровень параллелизма (через семафоры).
• Поддержка батч-обработки.
• Стриминговая передача данных между стадиями.
• Детерминированный порядок вывода.

---

Архитектура конвейера

emails → SelectUsers → SelectMessages → CheckSpam → CombineResults → output

Каждая стадия принимает канал на вход (in <-chan interface{})
и возвращает канал на выход (out chan interface{}).

---

Стадии обработки

1️⃣ SelectUsers

• in - string внутри interface{}. это имейлы юзеров
• out - отдает структурки User{}. это результат функции GetUser()
• особенности:
  • GetUser() выполняется 1 секунду. его можно вызывать параллельно для нескольких юзеров. это экономит время
  • у некоторых юзеров есть alias'ы(псевдонимы). то есть на вид это два разных имейла, но на самом деле это один и тот же юзер в базе. например "batman@mail.ru" - это алиас к "bruce.wayne@mail.ru". SelectUsers() должен отдавать в out только уникальных юзеров.

---

2️⃣ SelectMessages

• in - User{} внутри interface{} от SelectUsers()
• out - отдает MsgID. это айдишники писем юзеров - результат функции GetMessages()
• особенности:
  • GetMessages() выполняется 1 секунду. его тоже можно вызывать параллельно.
  • но GetMessages() позволяет использовать "батчи". то есть за раз в нее можно запихнуть не 1 юзера, а несколько. максимальное кол-во юзеров = 2. то есть если мы хотим селектнуть у 10ти юзеров письма, то это можно сделать за 5 вызовов GetMessages(). в тестах проверяется, что кол-во вызовов оптимальное

---

3️⃣ CheckSpam

• in - MsgID внутри interface{}. это айдишники писем
• out - MsgData{}. это структура с парой полей: id и факт того является ли письмо спамом. это результат работы HasSpam()
• особенности:
  • HasSpam() симулирует поход в сервис антиспама, чтоб проверить письмо на наличие спама. один запрос выполняется за 100мс. и у этого сервиса есть "антибрут" - его нельзя вызывать бесконтрольно в кучу потоков. если сделать к нему более 5 параллельных запросов, то он начнет возвращать ошибку и данные о наличии спама вы не получите.

---

4️⃣ CombineResults

• in - MsgData внутри interface{}
• out - string. это строки вида "<has_spam> <msg_id>", например "true 17696166526272393238"
• особенности:

  • CombineResults() ждет все результаты из in, а потом сортирует их по наличию спама и по msg_id. то есть пример вывода может быть такой:

      true 123
      true 5555
      true 5556
      false 140
      false 3000
      false 3005
    

---

Функция RunPipeline

Запускает цепочку стадий и последовательно связывает их каналы.
Каждая стадия работает в отдельной горутине и закрывает out после завершения.

RunPipeline(
  SelectUsers,
  SelectMessages,
  CheckSpam,
  CombineResults,
)

Тестирование

go test -v -race