В сервис из очереди приходят обновления записей. Нужно написать и логику
склейки, и цикл консьюмера, который читает очередь и пишет результат дальше.
Сервис держит несколько консьюмеров (по одному на партицию), и все они работают
с одним Processor.
type Record struct {
Key string
PublishedAt uint64
SeenAt uint64 // момент получения версии; пара (Key, SeenAt) уникальна
Body string
FirstSeenAt uint64 // заполняешь ты
}
type Processor interface {
Process(Record) (*Record, error)
}
// Source — входная очередь. Next блокируется до следующего сообщения;
// когда поток закончился, возвращает io.EOF.
type Source interface {
Next(ctx context.Context) (Record, error)
}
// Sink — выходная очередь.
type Sink interface {
Send(ctx context.Context, r Record) error
}
var ErrEmptyKey = errors.New("empty key")
func NewProcessor() Processor
func Run(ctx context.Context, src Source, dst Sink, p Processor) error
Требования к Processor
- Для каждого
Key склейка всех увиденных версий: Body и SeenAt — из версии
с максимальным SeenAt; PublishedAt — из версии с минимальным SeenAt;
FirstSeenAt — минимальный SeenAt.
- Обновления приходят не по порядку и с повторами. Если склейка не изменилась,
Process возвращает nil, nil.
- Пустой
Key → nil, ErrEmptyKey.
Process вызывают одновременно из многих горутин. Код проходит
go test -race, и итог не зависит от того, как легли сообщения по горутинам.
Требования к Run
- Читает
src до io.EOF и тогда возвращает nil.
- Каждое сообщение отдаёт в
p.Process. Если результат не nil, отправляет его
в dst. Если nil, ничего не пишет.
- Ошибка
Process (битое сообщение) не останавливает консьюмер: сообщение
пропускается, чтение продолжается.
- Ошибка
Next (кроме io.EOF) или Send останавливает Run, и он возвращает
ошибку, для которой errors.Is находит исходную.
- После отмены
ctx Run возвращает ctx.Err() и больше не читает
сообщения, даже если Source контекст игнорирует.
Пример
p := NewProcessor()
p.Process(Record{Key: "k", SeenAt: 20, PublishedAt: 2, Body: "new"})
// → {k, PublishedAt: 2, SeenAt: 20, Body: "new", FirstSeenAt: 20}
p.Process(Record{Key: "k", SeenAt: 10, PublishedAt: 1, Body: "old"})
// → {k, PublishedAt: 1, SeenAt: 20, Body: "new", FirstSeenAt: 10}
p.Process(Record{Key: "k", SeenAt: 10, PublishedAt: 1, Body: "old"})
// → nil: повтор
Устно: когда коммитить оффсет — до Process, после Send? Что будет при
падении между Send и коммитом? Куда девать битые сообщения?