Golang

8248

¿Qué estaban almacenando en caché? ¿Algún directorio?

/* Necesitamos transferir datos desde una fuente a un consumidor. La fuente entrega los datos en pequeños lotes (~decenas de registros), mientras que el consumidor funciona de manera más eficiente con lotes grandes. Un ejemplo real es la transferencia de datos desde colas tipo Kafka a la base de datos Clickhouse. Fuente: - Prácticamente infinita. - La fuente nunca devuelve más de MaxItems registros en una sola llamada a Next. - Dentro de una "sesión" (una llamada a la función Pipe), la fuente devuelve datos nuevos en cada llamada a Next. - Sin embargo, tras reiniciar, la fuente comenzará desde la posición "confirmada" anterior, indicada por cookie. Por lo tanto, *cada* valor de cookie que devuelve Next, después de guardar los datos en el receptor, debe ser confirmado con una llamada a Commit, en el mismo orden en que fueron devueltos por Next. Receptor: - No puede procesar más de MaxItems a la vez. Nivel básico: Se requiere implementar la función func Pipe(p Producer, c Consumer) error que lee datos de la fuente, los agrupa en un buffer de tamaño no mayor a MaxItems y los guarda en el receptor, y luego confirma el progreso en la fuente. */ const MaxItems = 9999 type Producer interface { // Next devuelve: // - un lote de items para procesar // - una cookie para confirmar cuando se complete el procesamiento // - error Next() (items []any, cookie int, err error) // Commit se usa para marcar el lote de datos como procesado Commit(cookie int) error } type Consumer interface { Process(items []any) error } func Pipe(p Producer, c Consumer) error { var buf []any var cookies []int for { items, cookie, err := p.Next() if err != nil { return err } buf = append(buf, items...) cookies = append(cookies, cookie) if len(buf) >= MaxItems { if err := c.Process(buf); err != nil { return err } for _, c := range cookies { if err := p.Commit(c); err != nil { return err } } buf = buf[:0] cookies = nil } } if len(buf) > 0 { if err := c.Process(buf); err != nil { return err } for _, c := range cookies { if err := p.Commit(c); err != nil { return err } } } return nil }