mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-12-15 16:30:55 +01:00
lib/protoparser: stop read when callback error (#1380)
This commit is contained in:
parent
5f91a701fa
commit
2ea187e801
@ -56,7 +56,7 @@ func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback f
|
||||
|
||||
func (ctx *streamContext) Read() bool {
|
||||
readCalls.Inc()
|
||||
if ctx.err != nil {
|
||||
if ctx.err != nil || ctx.CallbackError() != nil {
|
||||
return false
|
||||
}
|
||||
ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
|
||||
@ -77,7 +77,7 @@ type streamContext struct {
|
||||
err error
|
||||
|
||||
wg sync.WaitGroup
|
||||
callbackErrLock sync.Mutex
|
||||
callbackErrLock sync.RWMutex
|
||||
callbackErr error
|
||||
}
|
||||
|
||||
@ -88,6 +88,13 @@ func (ctx *streamContext) Error() error {
|
||||
return ctx.err
|
||||
}
|
||||
|
||||
func (ctx *streamContext) CallbackError() error {
|
||||
ctx.callbackErrLock.RLock()
|
||||
callbackErr := ctx.callbackErr
|
||||
ctx.callbackErrLock.RUnlock()
|
||||
return callbackErr
|
||||
}
|
||||
|
||||
func (ctx *streamContext) reset() {
|
||||
ctx.br.Reset(nil)
|
||||
ctx.reqBuf = ctx.reqBuf[:0]
|
||||
|
Loading…
Reference in New Issue
Block a user