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