/*
Мо бояд маълумотҳоро аз манбаъ ба истеъмолкунанда гузаронид.
Манбаъ маълумотҳоро дар қисмҳои хурд (~даҳ сабт) медиҳад, дар ҳоле ки истеъмолкунанда бо қисмҳои калонтар (~ҳазор сабт) беҳтар кор мекунад.
Масалан, интиқоли маълумот аз навъҳои 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 бармегардонад:
// - як партия аз ашё барои коркард
// - cookie барои тасдиқ кардан пас аз анҷоми кор
// - хато
Next() (items []any, cookie int, err error)
// Commit барои нишон додани он, ки партияи маълумот коркард шудааст
Commit(cookie int) error
}
type Consumer interface {
Process(items []any) error
}
func Pipe(p Producer, c Consumer) error {
// TODO
}
