Module: Fluent::Plugin::Extractor

Defined in:
lib/fluent/plugin/extractor.rb

Defined Under Namespace

Classes: SizeLimitError

Constant Summary collapse

BYTES_TO_READ = 
64 * 1024
INFLATE_BYTES_TO_READ = 
1024
ZSTD_BYTES_TO_READ =

Zstd::StreamReader#read takes a number of compressed bytes, unlike Zlib::GzipReader#read which takes decompressed bytes

1024

Class Method Summary collapse

Class Method Details

.decompress_deflate(compressed_data, limit:) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/fluent/plugin/extractor.rb', line 72

def self.decompress_deflate(compressed_data, limit:)
  io = StringIO.new(compressed_data)
  out = ''
  begin
    zstream = Zlib::Inflate.new
    while (chunk = io.read(INFLATE_BYTES_TO_READ))
      out << zstream.inflate(chunk)
      if out.bytesize > limit
        raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
      end
    end
    out << zstream.finish
    if out.bytesize > limit
      raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
    end
  ensure
    zstream&.close
  end
  out
end

.decompress_gzip(compressed_data, limit:) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/fluent/plugin/extractor.rb', line 33

def self.decompress_gzip(compressed_data, limit:)
  io = StringIO.new(compressed_data)
  out = ''
  loop do
    reader = Zlib::GzipReader.new(io)
    while (chunk = reader.read(BYTES_TO_READ))
      out << chunk
      if out.bytesize > limit
        raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
      end
    end

    unused = reader.unused
    reader.finish
    unless unused.nil?
      adjust = unused.length
      io.pos -= adjust
    end
    break if io.eof?
  end
  out
end

.decompress_zstd(compressed_data, limit:) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/fluent/plugin/extractor.rb', line 56

def self.decompress_zstd(compressed_data, limit:)
  io = StringIO.new(compressed_data)
  reader = Zstd::StreamReader.new(io)
  out = ''
  loop do
    out << reader.read(ZSTD_BYTES_TO_READ)
    if out.bytesize > limit
      raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
    end

    # Zstd::StreamReader doesn't provide unused data, so we have to manually adjust the position
    break if io.eof?
  end
  out
end

.io_decompress_gzip(input, output, limit:) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/fluent/plugin/extractor.rb', line 93

def self.io_decompress_gzip(input, output, limit:)
  size = 0
  loop do
    reader = Zlib::GzipReader.new(input)
    while (chunk = reader.read(BYTES_TO_READ))
      size += chunk.bytesize
      if size > limit
        raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
      end
      output.write(chunk)
    end
    unused = reader.unused
    reader.finish
    unless unused.nil?
      adjust = unused.length
      input.pos -= adjust
    end
    break if input.eof?
  end
  output
end

.io_decompress_zstd(input, output, limit:) ⇒ Object 



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/fluent/plugin/extractor.rb', line 115

def self.io_decompress_zstd(input, output, limit:)
  reader = Zstd::StreamReader.new(input)
  size = 0
  loop do
    chunk = reader.read(ZSTD_BYTES_TO_READ)
    size += chunk.bytesize
    if size > limit
      raise SizeLimitError, "Decompressed data exceeds limit of #{limit} bytes"
    end
    output.write(chunk)
    # Zstd::StreamReader doesn't provide unused data, so we have to manually adjust the position
    break if input.eof?
  end
  output
end