fix(inputs.cloud_pubsub): Fix gzip decompression (#13238)
This commit is contained in:
parent
1b8339ace8
commit
43048aad8c
|
|
@ -62,9 +62,10 @@ type PubSub struct {
|
|||
wg *sync.WaitGroup
|
||||
acc telegraf.TrackingAccumulator
|
||||
|
||||
undelivered map[telegraf.TrackingID]message
|
||||
sem semaphore
|
||||
decoder internal.ContentDecoder
|
||||
undelivered map[telegraf.TrackingID]message
|
||||
sem semaphore
|
||||
decoder internal.ContentDecoder
|
||||
decoderMutex sync.Mutex
|
||||
}
|
||||
|
||||
func (*PubSub) SampleConfig() string {
|
||||
|
|
@ -216,10 +217,13 @@ func (ps *PubSub) decompressData(data []byte) ([]byte, error) {
|
|||
return data, nil
|
||||
}
|
||||
|
||||
ps.decoderMutex.Lock()
|
||||
data, err := ps.decoder.Decode(data, int64(ps.MaxDecompressionSize))
|
||||
if err != nil {
|
||||
ps.decoderMutex.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
ps.decoderMutex.Unlock()
|
||||
|
||||
decompressedData := make([]byte, len(data))
|
||||
copy(decompressedData, data)
|
||||
|
|
|
|||
Loading…
Reference in New Issue