Нам нужно передать данные из некоторого источника некоторому потребителю.

Условие

/* Нам нужно передать данные из некоторого источника некоторому потребителю. При этом источник отдает данные небольшими пачками (~ десятки записей), а потребитель оптимальнее работает с крупными батчами (~тысячи записей). Реальный пример - поставка данных из очередей типа Kafka в базу Clickhouse.

Источник:

• Условно бесконечный.
• Источник никогда не возвращает более MaxItems записей за один вызов Next.
• В рамках одной "сессии" (одного вызова функции Pipe) источник каждый раз возвращает новые данные на каждый вызов Next.
• Однако, после перезапуска источник начнет с прошлой "подтвержденной" позиции, задаваемой cookie. Поэтому каждое значение cookie, которое вернул вызов Next, после сохранения данных в приемнике, должно быть фиксировано вызовом Commit, причем строго в той же последовательности, в которой их вернул Next

Приемник:

• Не может обработать более MaxItems за один раз.

Требуется реализовать функцию func Pipe(p Producer, c Consumer) error которая читает данные из источника, группирует их в буфер размером не более MaxItems и сохраняет в приёмник, после чего фиксирует прогресс в источнике. */

const MaxItems = 9999

type Producer interface { // Next returns: // - batch of items to be processed // - cookie to be commited when processing is done // - error Next() (items []any, cookie int, err error) // Commit is used to mark data batch as processed Commit(cookie int) error }

type Consumer interface { Process(items []any) error }

func Pipe(p Producer, c Consumer) error { // TODO }

Ответ

Накапливайте записи вместе с cookies, сбрасывайте буфер до переполнения, затем подтверждайте каждый cookie строго по порядку. Commit возможен только после успешного Process соответствующих данных.

В этом варианте данные и cookie считаются действительными только при err == nil; условие не задаёт семантику частичного результата с ошибкой.

func Pipe(p Producer, c Consumer) error {
    buffer := make([]any, 0, MaxItems)
    cookies := make([]int, 0)
    flush := func() error {
        if len(buffer) > 0 {
            if err := c.Process(buffer); err != nil { return err }
        }
        for _, cookie := range cookies {
            if err := p.Commit(cookie); err != nil { return err }
        }
        for i := range buffer { buffer[i] = nil }
        buffer = buffer[:0]
        cookies = cookies[:0]
        return nil
    }
    for {
        items, cookie, err := p.Next()
        if err != nil {
            if flushErr := flush(); flushErr != nil { return flushErr }
            return err
        }
        if len(buffer)+len(items) > MaxItems {
            if err := flush(); err != nil { return err }
        }
        buffer = append(buffer, items...)
        cookies = append(cookies, cookie)
        if len(buffer) == MaxItems || len(items) == 0 {
            if err := flush(); err != nil { return err }
        }
    }
}

Process синхронно использует слайс и не удерживает его после возврата. Пустая порция тоже подтверждается, но только после предшествующих cookies. Сбой между Process и Commit даёт повтор после рестарта: приёмник должен быть идемпотентным. Для таймерного flush при блокирующем Next интерфейс нужно расширить отменой или отдельным источником событий.