Class: HttpStreamingClient::Decoders::GZip

Inherits:
Object
  • Object
show all
Defined in:
lib/http_streaming_client/decoders/gzip.rb

Defined Under Namespace

Classes: GZipBufferIO

Constant Summary collapse

GZIP_READER_MIN_BUF_SIZE =

MRI: pass at least 2k bytes to GzipReader to avoid zlib EOF

2048

Instance Method Summary collapse

Constructor Details

#initialize(&packet_callback) ⇒ GZip

Returns a new instance of GZip.



54
55
56
57
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 54

def initialize(&packet_callback)
  logger.debug "GZip:initialize"
  @packet_callback = packet_callback
end

Instance Method Details

#<<(compressed_packet) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 59

def <<(compressed_packet)
  return unless compressed_packet && compressed_packet.size > 0
  @buf ||= GZipBufferIO.new
  @buf << compressed_packet
  @gzip ||= Zlib::GzipReader.new @buf

  # pass at least GZIP_READER_MIN_BUF_SIZE bytes to GzipReader to avoid zlib EOF
  while @buf.size > GZIP_READER_MIN_BUF_SIZE do
    decompressed_packet = nonblock_readline(@gzip)
    #logger.debug "GZip:<<:decompressed_packet:#{decompressed_packet}"
    break if decompressed_packet.nil?
    process_decompressed_packet(decompressed_packet)
  end
end

#close ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 74

def close
  logger.debug "GZip:close"
  return if @buf.size == 0

  decompressed_packet = ""
  begin
    @gzip ||= Zlib::GzipReader.new @buf

    while true do
      decompressed_packet = nonblock_readline(@gzip, true)
      logger.debug "GZip:close:decompressed_packet:#{decompressed_packet}"
      break if decompressed_packet.nil? or decompressed_packet.size == 0
      process_decompressed_packet(decompressed_packet)
    end

  rescue Zlib::Error => e
    raise HttpStreamingClient::DecoderError.new(e.message)
  end
end

#logger ⇒ Object 



50
51
52
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 50

def logger
  HttpStreamingClient.logger
end

#nonblock_readline(io, on_close = false) ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 98

def nonblock_readline(io, on_close = false)
  @line_buffer ||= ""
  ch = nil
  begin
    while ch = io.getc
      @line_buffer += ch
      if ch == "\n" then
 result = @line_buffer
 @line_buffer = ""
 return result
      end
    end
  rescue Zlib::GzipFile::Error
    # this is raised on EOF by ZLib in MRI
    # if we get here via a call to close(), then we want to return the line_buffer
    # if we get here via any other path, we want to return nil to signal temporary EOF and leave the partial line_buffer contents in place.
    logger.debug "Gzip:nonblock_readline:Zlib::GzipFile::Error:line_buffer.size:#{@line_buffer.size}"
    if on_close then
      result = @line_buffer
      @line_buffer = ""
      return result
    end
    return nil
  rescue IOError
    # this is raised on EOF by ZLib in JRuby, return nil to indicate EOF and leave partial line in the buffer
    # if we get here via a call to close(), then we want to return the line_buffer
    # if we get here via any other path, we want to return nil to signal temporary EOF and leave the partial line_buffer contents in place.
    logger.debug "Gzip:nonblock_readline:IOError:line_buffer.size:#{@line_buffer.size}"
    if on_close then
      result = @line_buffer
      @line_buffer = ""
      return result
    end
    return nil
  rescue => e
    logger.debug "Gzip:nonblock_readline:error received:#{e.class}:#{e}"
    raise e
  end
    
  if on_close then
    result = @line_buffer
    @line_buffer = ""
    return result
  end

end

#size ⇒ Object 



94
95
96
 
# File 'lib/http_streaming_client/decoders/gzip.rb', line 94

def size
  @buf.size
end