Message batching queue

This package very useful for organizing batch messages.
Can help you create batch inserts to a database for example. Thread safe and well tested.

// create new queue objectbatch:=mb.New((any)(nil), 0)
// add new message to the queuebatch.Add(msg)
// wait until anybody add message/messages// will return the slice of all queued messages. ([]T)messages:=batch.Wait(ctx)
// wait until count of messages will be more than 10// if we have more than 100 messages, will be returned only 100messages:=batch.NewCond().WithMin(10).WithMax(100).Wait(ctx)
// when we have 0 messages returned that means the queue is closed.iflen(messages) ==0 {
return
}
// close queue// if the queue has messages all receivers will get remaining data.batch.Close()

Docs

https://godoc.org/github.com/cheggaaa/mb/v3

Installation

go get -u github.com/cheggaaa/mb/v3

Example

package main
import (
"context""fmt""time""github.com/cheggaaa/mb/v3"
)
funcmain() {
ctx:=context.Background()
// create the queue with 10 items capacityq:= mb.New[int](10)
// create the channel for showing when all work will be donedone:=make(chanbool)
// start two workersgoworker(ctx, "first", q, done)
goworker(ctx, "second", q, done)
// start two publishersgopublisher(ctx, "first", q)
gopublisher(ctx, "second", q)
// give time to worktime.Sleep(time.Second)
// close the queueq.Close()
// and wait until all sent messages will be processedfori:=0; i<2; i++ {
<-done
}
}
funcpublisher(ctx context.Context, namestring, q*mb.MB[string]) {
fmt.Printf("Publisher %s: started\n", name)
variintfor {
// will sending name and countermsg:=fmt.Sprintf("%s - %d", name, i)
// addiferr:=q.Add(ctx, msg); err!=nil {
// non-nil err mean that queue is closedbreak
}
// 10 messages per secondtime.Sleep(time.Second/10)
i++
}
fmt.Printf("Publisher %s: closed\n", name)
}
funcworker(ctx context.Context, namestring, q*mb.MB[string], donechanbool) {
fmt.Printf("Worker %s: started\n", name)
for {
// getting messagesmsgs, err:=q.Wait(ctx)
iferr!=nil {
break
}
msgsForPrint:=""for_, msg:=rangemsgs {
msgsForPrint+=fmt.Sprintf("\t%s\n", msg)
}
fmt.Printf("Worker %s: %d messages received\n%s", name, len(msgs), msgsForPrint)
// doing working, for example, send messages to remote servertime.Sleep(time.Second/3)
}
fmt.Printf("Worker %s: closed\n", name)
done<-true
}