Golang

8248

Cosa stavano memorizzando nella cache? Qualche directory?

/* Dobbiamo trasferire dati da una fonte a un consumatore. La fonte fornisce i dati in piccoli batch (~decine di record), mentre il consumatore funziona in modo più efficiente con batch più grandi. Un esempio reale è il trasferimento di dati da code di tipo Kafka al database Clickhouse. Fonte: - Quasi infinita. - La fonte non restituisce mai più di MaxItems record in una singola chiamata a Next. - All'interno di una "sessione" (una chiamata alla funzione Pipe), la fonte restituisce nuovi dati ad ogni Next. - Tuttavia, dopo un riavvio, la fonte ricomincerà dalla posizione "confermata" precedente, indicata da cookie. Pertanto, *ogni* valore di cookie restituito da Next, dopo aver salvato i dati nel ricevitore, deve essere confermato con una chiamata a Commit, nello stesso ordine in cui sono stati restituiti da Next. Ricevitore: - Non può elaborare più di MaxItems alla volta. Livello di base: È necessario implementare la funzione func Pipe(p Producer, c Consumer) error che legge i dati dalla fonte, li raggruppa in un buffer di dimensione non superiore a MaxItems e li salva nel ricevitore, e poi conferma il progresso nella fonte. */ const MaxItems = 9999 type Producer interface { // Next restituisce: // - batch di elementi da elaborare // - cookie da confermare al termine dell'elaborazione // - errore Next() (items []any, cookie int, err error) // Commit viene usato per marcare il batch di dati come processato 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 }