upstream/icingadb
1
0
Fork 0
mirror of https://github.com/Icinga/icingadb.git synced 2026-05-28 04:35:54 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

DB#NamedBulkExec(): optionally inform about successfully inserted rows

Browse source
This commit is contained in:
Alexander A. Klimov 2021-03-02 17:00:42 +01:00
parent b13d2c3cd7
commit df0124de09
1 changed files with 13 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
pkg/icingadb/db.go
View file

@ -159,7 +159,8 @@ func (db DB) BulkExec(ctx context.Context, query string, count int, concurrent i
}
func (db DB) NamedBulkExec(
ctx context.Context, query string, count int, concurrent int, arg chan contracts.Entity,
ctx context.Context, query string, count int, concurrent int,
arg chan contracts.Entity, succeeded chan<- contracts.Entity,
) error {
var cnt com.Counter
g, ctx := errgroup.WithContext(ctx)
@ -206,6 +207,16 @@ func (db DB) NamedBulkExec(
cnt.Add(uint64(len(b)))
if succeeded != nil {
for _, row := range b {
select {
case <-ctx.Done():
return ctx.Err()
case succeeded <- row:
}
}
}
return nil
},
IsRetryable,
@ -354,7 +365,7 @@ func (db DB) Create(ctx context.Context, entities <-chan contracts.Entity) error
}
}()
return db.NamedBulkExec(ctx, db.BuildInsertStmt(entity), 1<<15/len(db.BuildColumns(entity)), 1<<3, inserts)
return db.NamedBulkExec(ctx, db.BuildInsertStmt(entity), 1<<15/len(db.BuildColumns(entity)), 1<<3, inserts, nil)
}
func (db DB) Update(ctx context.Context, entities <-chan contracts.Entity) error {