Class: Baykit::BayServer::Tours::TourRes

Inherits:
Object
  • Object
show all
Includes:
Agent::Transporter, Docker, Baykit::BayServer::Taxi, Baykit::BayServer::Tours, Util, Util::Reusable
Defined in:
lib/baykit/bayserver/tours/tour_res.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(tur) ⇒ TourRes

Returns a new instance of TourRes.



51
52
53
54
55
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 51

def initialize(tur)
  @headers = Headers.new()
  @tour = tur
  @buffer_size = BayServer.harbor.tour_buffer_size
end

Instance Attribute Details

#available ⇒ Object (readonly)

Returns the value of attribute available.



34
35
36
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 34

def available
  @available
end

#buffer_size ⇒ Object (readonly)

Returns the value of attribute buffer_size.



49
50
51
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 49

def buffer_size
  @buffer_size
end

#bytes_consumed ⇒ Object (readonly)

Returns the value of attribute bytes_consumed.



47
48
49
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 47

def bytes_consumed
  @bytes_consumed
end

#bytes_limit ⇒ Object (readonly)

Returns the value of attribute bytes_limit.



48
49
50
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 48

def bytes_limit
  @bytes_limit
end

#bytes_posted ⇒ Object (readonly)

Returns the value of attribute bytes_posted.



46
47
48
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 46

def bytes_posted
  @bytes_posted
end

#can_compress ⇒ Object (readonly)

Response Content info



43
44
45
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 43

def can_compress
  @can_compress
end

#charset ⇒ Object

Returns the value of attribute charset.



33
34
35
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 33

def charset
  @charset
end

#compressor ⇒ Object (readonly)

Returns the value of attribute compressor.



44
45
46
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 44

def compressor
  @compressor
end

#header_sent ⇒ Object

Returns the value of attribute header_sent.



37
38
39
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 37

def header_sent
  @header_sent
end

#headers ⇒ Object (readonly)

Response Header info



32
33
34
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 32

def headers
  @headers
end

#res_consume_listener ⇒ Object (readonly)

Returns the value of attribute res_consume_listener.



35
36
37
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 35

def res_consume_listener
  @res_consume_listener
end

#tour ⇒ Object (readonly)

Returns the value of attribute tour.



27
28
29
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 27

def tour
  @tour
end

#yacht ⇒ Object (readonly)

Returns the value of attribute yacht.



38
39
40
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 38

def yacht
  @yacht
end

Instance Method Details

#buffer_available ⇒ Object 



407
408
409
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 407

def buffer_available()
  return @bytes_posted - @bytes_consumed < @buffer_size
end

#consumed(check_id, length) ⇒ Object 



313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 313

def consumed(check_id, length)
  @tour.check_tour_id(check_id)
  if @res_consume_listener == nil
    raise Sink.new("Response consume listener is null")
  end

  @bytes_consumed += length

  BayLog.debug("%s resConsumed: len=%d posted=%d consumed=%d limit=%d",
               @tour, length, @bytes_posted, @bytes_consumed, @bytes_limit)

  resume = false
  old_available = @available
  if buffer_available()
    @available = true
  end
  if !old_available && @available
    BayLog.debug("%s response available (^o^): posted=%d consumed=%d", self,  @bytes_posted, @bytes_consumed)
    resume = true
  end

  if @tour.running?
    @res_consume_listener.call(length, resume)
  end
end

#end_res_content(chk_tour_id) ⇒ Object 



262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 262

def end_res_content(chk_tour_id)
  @tour.check_tour_id(chk_tour_id)

  BayLog.debug("%s end ResContent", self)
  if @tour.ended?
    BayLog.debug("%s Tour is already ended (Ignore).", self)
    return
  end

  if !@tour.zombie? && @tour.city != nil
    @tour.city.log(@tour)
  end

  # send end message
  if @can_compress
    get_compressor().finish()
  end


  # Done listener
  tour_returned = false
  done_lis = Proc.new() do
    @tour.check_tour_id(chk_tour_id)
    @tour.ship.return_tour(@tour)
    tour_returned = true
  end

  begin
    if @tour.zombie? || @tour.aborted?
      # Don't send peer any data. Do nothing
      BayLog.debug("%s Aborted or zombie tour. do nothing: %s state=%s", self, @tour, @tour.state)
      done_lis.call()
    else
      begin
        @tour.ship.send_end_tour(@tour.ship_id, @tour, &done_lis)
      rescue IOError => e
        BayLog.debug("%s Error on sending end tour", self)
        done_lis.call()
        raise e
      end
    end
  ensure
    # If tour is returned, we cannot change its state because
    # it will become uninitialized.
    BayLog.debug("%s tur#%d is returned: %s", self, chk_tour_id, tour_returned)
    if !tour_returned
      @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ENDED)
    end
  end
end

#get_compressor ⇒ Object 



387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 387

def get_compressor()
  if @compressor == nil
    sip_id = @tour.ship.ship_id
    tur_id = @tour.tour_id
    gz_callback = lambda do |new_buf, new_ofs, new_len, &lis|
      begin
        @tour.ship.send_res_content(sip_id, @tour, new_buf, new_ofs, new_len, &lis)
      rescue IOError => e
        @tour.change_state(tur_id, Tour::TourState::ABORTED)
        raise e
      end
    end

    @compressor = GzipCompressor.new(gz_callback)
  end

  return @compressor
end

#init ⇒ Object 



57
58
59
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 57

def init()
  @yacht = SendFileYacht.new()
end

#reset ⇒ Object

Implements Reusable



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 69

def reset()
  @charset = nil
  @header_sent = false
  if @yacht != nil
    @yacht.reset()
  end

  @available = false
  @res_consume_listener = nil

  @can_compress = false
  @compressor = nil
  @headers.clear()
  @bytes_posted = 0
  @bytes_consumed = 0
  @bytes_limit = 0
  @tour_returned = false
end

#send_error(chk_tour_id, status = HttpStatus::INTERNAL_SERVER_ERROR, msg = "", err = nil) ⇒ Object 



347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 347

def send_error(chk_tour_id, status=HttpStatus::INTERNAL_SERVER_ERROR, msg="", err=nil)
  @tour.check_tour_id(chk_tour_id)
  #BayLog.debug "#{self} Tur: Send Error status=#{status} msg=#{msg}"

  if @tour.zombie?
    return
  end


  if err.instance_of?(HttpException)
    status = err.status
    msg = err.message
  end

  if @header_sent
    BayLog.debug("Try to send error after response header is sent (Ignore)")
    BayLog.debug("%s: status=%d, message=%s", self, status, msg)
    if err != nil
      BayLog.error_e(err);
    end
  else
    set_consume_listener(&ContentConsumeListener::DEV_NULL)

    if @tour.zombie? || @tour.aborted?
      # Don't send peer any data. Do nothing
      BayLog.debug("%s Aborted or zombie tour. do nothing: %s state=%s", self, @tour, @tour.state)
    else
      begin
        @tour.ship.send_error(@tour.ship_id, @tour, status, msg, err)
      rescue IOError => e
        BayLog.debug_e(e, "%s Error on sending error", self)
        @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ABORTED)
      end
      @header_sent = true
    end
  end

  end_res_content(chk_tour_id)
end

#send_headers(chk_tour_id) ⇒ Object

other methods



92
93
94
95
96
97
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 92

def send_headers(chk_tour_id)
  @tour.check_tour_id(chk_tour_id)

  if @tour.zombie?
    return
  end

  if @header_sent
    return
  end

  @bytes_limit = @headers.content_length()
  BayLog.debug("%s send_headers content length: %s", self, @bytes_limit)

  # Compress check
  if BayServer.harbor.gzip_comp &&
    @headers.contains(Headers::CONTENT_TYPE) &&
    @headers.content_type().downcase().start_with?("text/") &&
    !@headers.contains(Headers::CONTENT_ENCODING)

    enc = @tour.req.headers.get(Headers::ACCEPT_ENCODING)

    if enc != nil
      enc.split(",").each do |tkn|
        if tkn.strip().casecmp?("gzip")
          @can_compress = true
          @headers.set(Headers::CONTENT_ENCODING, "gzip")
          @headers.remove(Headers::CONTENT_LENGTH)
          break
        end
      end
    end
  end

  begin
    handled = false
    if !@tour.error_handling && @tour.res.headers.status >= 400
      trouble = BayServer.harbor.trouble
      if trouble != nil
        cmd = trouble.find(tur.res.headers.status)
        if cmd != nil
          err_tour = get_error_tour
          err_tour.req.uri = cmd.target
          @tour.req.headers.copy_to(err_tour.req.headers)
          @tour.res.headers.copy_to(err_tour.res.headers)
          err_tour.req.remote_port = @tour.req.remote_port
          err_tour.req.remote_address = @tour.req.remote_address
          err_tour.req.server_address = @tour.req.server_address
          err_tour.req.server_port = @tour.req.server_port
          err_tour.req.server_name = @tour.req.server_name
          err_tour.res.header_sent = @tour.res.header_sent
          @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ZOMBIE)
          case cmd.method
          when :GUIDE
            err_tour.go
          when :TEXT
            @protocol_handler.send_headers(err_tour)
            data = cmd.target
            err_tour.res.send_res_content(Tour::TOUR_ID_NOCHECK, data, 0, data.length)
            err_tour.end_res_content(Tour::TOUR_ID_NOCHECK)
          when :REROUTE
            err_tour.res.send_http_exception(Tour::TOUR_ID_NOCHECK, HttpException.moved_temp(cmd.target))
          end
          handled = true
        end
      end
    end

    if !handled
      @tour.ship.send_headers(@tour.ship_id, @tour)
    end
  rescue IOError => e
    BayLog.debug_e(e, "%s abort: %s", @tour, e)
    @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ABORTED)
    raise e
  ensure
    @header_sent = true
  end

end

#send_http_exception(chk_tour_id, http_ex) ⇒ Object 



339
340
341
342
343
344
345
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 339

def send_http_exception(chk_tour_id, http_ex)
  if http_ex.status == HttpStatus::MOVED_TEMPORARILY || http_ex.status == HttpStatus::MOVED_PERMANENTLY
    send_redirect(chk_tour_id, http_ex.status, http_ex.location)
  else
    send_error(chk_tour_id, http_ex.status, http_ex.message, http_ex)
  end
end

#send_redirect(chk_tour_id, status, location) ⇒ Object 



173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 173

def send_redirect(chk_tour_id, status, location)
  @tour.check_tour_id(chk_tour_id)

  if @header_sent
    BayLog.error("Try to redirect after response header is sent (Ignore)")
  else
    set_consume_listener(&ContentConsumeListener::DEV_NULL)
    begin
      @tour.ship.send_redirect(@tour.ship_id, @tour, status, location)
    rescue IOError => e
      @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ABORTED)
      raise e
    ensure
      @header_sent = true
      end_content(chk_tour_id)
    end
  end

end

#send_res_content(chk_tour_id, buf, ofs, len) ⇒ Object 



200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 200

def send_res_content(chk_tour_id, buf, ofs, len)
  @tour.check_tour_id(chk_tour_id)
  BayLog.debug("%s sendContent len=%d", @tour, len)

  # Done listener
  done_lis = Proc.new() do
    consumed(chk_tour_id, len);
  end

  if @tour.zombie?
    BayLog.debug("%s zombie return", self)
    done_lis.call()
    return true
  end

  if !@header_sent
    raise Sink.new("Header not sent")
  end


  if @res_consume_listener == nil
    raise Sink.new("Response consume listener is null")
  end

  @bytes_posted += len
  BayLog.debug("%s posted res content len=%d posted=%d limit=%d consumed=%d",
  @tour, len, @bytes_posted, @bytes_limit, @bytes_consumed)
  if @bytes_limit > 0 && @bytes_limit < self.bytes_posted
    raise ProtocolException.new("Post data exceed content-length: " + @bytes_posted + "/" + @bytes_limit)
  end

  if @tour.zombie? || @tour.aborted?
    # Don't send peer any data
    BayLog::debug("%s Aborted or zombie tour. do nothing: %s state=%s", self, @tour, @tour.state)
    done_lis.call()
  else
    if @can_compress
      get_compressor().compress(buf, ofs, len, &done_lis)
    else
      begin
        @tour.ship.send_res_content(@tour.ship_id, @tour, buf, ofs, len, &done_lis)
      rescue IOError => e
        done_lis.call()
        @tour.change_state(Tour::TOUR_ID_NOCHECK, Tour::TourState::ABORTED)
        raise e
      end
    end
  end


  old_available = @available
  if !buffer_available()
    @available = false
  end
  if old_available && !@available
    BayLog.debug("%s response unavailable (_ _): posted=%d consumed=%d (buffer=%d)",
                 self, @bytes_posted, @bytes_consumed, @buffer_size)
  end

  return @available
end

#set_consume_listener(&listener) ⇒ Object 



193
194
195
196
197
198
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 193

def set_consume_listener(&listener)
  @res_consume_listener = listener
  @bytes_consumed = 0
  @bytes_posted = 0
  @available = true
end

#to_s ⇒ Object 



61
62
63
 
# File 'lib/baykit/bayserver/tours/tour_res.rb', line 61

def to_s()
  return @tour.to_s()
end