Class: Subscriber::LongPollClient

Inherits:
Client
  • Object
show all
Includes:
Celluloid::IO
Defined in:
lib/nchan_tools/pubsub.rb

Defined Under Namespace

Classes: HTTP2Bundle, HTTPBundle

Instance Attribute Summary collapse

Attributes inherited from Client

#concurrency

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Client

#handle_bundle_error, inherited, lookup, #poke, #provides_msgid?, unique_aliases

Constructor Details

#initialize(subscr, opt = {}) ⇒ LongPollClient

Returns a new instance of LongPollClient.



980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
 
# File 'lib/nchan_tools/pubsub.rb', line 980

def initialize(subscr, opt={})
  super
  @last_modified, @etag, @timeout = opt[:last_modified], opt[:etag], opt[:timeout].to_i || 10
  @connect_timeout = opt[:connect_timeout]
  @subscriber=subscr
  @url=subscr.url
  @concurrency=opt[:concurrency] || opt[:clients] || 1
  @gzip=opt[:gzip]
  @retry_delay=opt[:retry_delay]
  @nomsg=opt[:nomsg]
  @bundles={}
  @body_buf=""
  @extra_headers = opt[:extra_headers]
  @verbose=opt[:verbose]
  @http2=opt[:http2] || opt[:h2]
end

Instance Attribute Details

#timeout ⇒ Object

Returns the value of attribute timeout.



979
980
981
 
# File 'lib/nchan_tools/pubsub.rb', line 979

def timeout
  @timeout
end

Class Method Details

.aliases ⇒ Object 



716
717
718
 
# File 'lib/nchan_tools/pubsub.rb', line 716

def self.aliases
  [:longpoll]
end

Instance Method Details

#close(bundle) ⇒ Object 



1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
 
# File 'lib/nchan_tools/pubsub.rb', line 1121

def close(bundle)
  if bundle
    bundle.done=true
    bundle.sock.close unless bundle.sock.closed?
    @bundles.delete bundle
  end
  @connected -= 1
  if @connected <= 0
    @cooked.signal true
  end
end

#error(*args) ⇒ Object 



720
721
722
723
 
# File 'lib/nchan_tools/pubsub.rb', line 720

def error(*args)
  @error_what||= ["#{@http2 ? "HTTP/2" : "HTTP"} Request"]
  super
end

#listen(bundle) ⇒ Object 



1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
 
# File 'lib/nchan_tools/pubsub.rb', line 1109

def listen(bundle)
  loop do
    begin
      return false if bundle.read == false
    rescue EOFError
      @subscriber.on_failure error(0, "Server Closed Connection"), bundle
      close bundle
      return false
    end
  end
end

#new_bundle(uri, opt = {}) ⇒ Object 



1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
 
# File 'lib/nchan_tools/pubsub.rb', line 1052

def new_bundle(uri, opt={})
  opt[:headers]||={}
  if @extra_headers
    opt[:headers].merge! @extra_headers
  end
  if @gzip
    opt[:headers]["Accept-Encoding"]="gzip, deflate"
  end
  b=(@http2 ? HTTP2Bundle : HTTPBundle).new(uri, opt)
  b.on_error do |msg, err|
    handle_bundle_error b, msg, err
  end
  b.verbose=@verbose
  setup_bundle b
  b
end

#request_code_ok(code, bundle) ⇒ Object 



1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
 
# File 'lib/nchan_tools/pubsub.rb', line 1032

def request_code_ok(code, bundle)
  if code != 200
    if code == 304 || code == 408
      @subscriber.on_failure error(code, "", bundle)
      @subscriber.finished+=1
      close bundle
    elsif @subscriber.on_failure(error(code, "", bundle)) == false
      @subscriber.finished+=1
      close bundle
    else
      Celluloid.sleep @retry_delay if @retry_delay
      bundle.send_GET
    end
    false
  else
    @timer.reset if @timer
    true
  end
end

#run(was_success = nil) ⇒ Object 



1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
 
# File 'lib/nchan_tools/pubsub.rb', line 1005

def run(was_success = nil)
  uri = URI.parse_possibly_unix_socket(@url)
  uri.port||= uri.scheme.match(/^(ws|http)$/) ? 80 : 443
  @cooked=Celluloid::Condition.new
  @connected = @concurrency
  @notready = @concurrency
  @timer.cancel if @timer
  if @timeout
    @timer = after(@timeout) do 
      stop "Timeout"
    end
  end
  @concurrency.times do |i|
    begin
      bundle = new_bundle(uri, id: i, useragent: "pubsub.rb #{self.class.name} #{@use_http2 ? "(http/2)" : ""} ##{i}", logger: @logger)
    rescue SystemCallError => e
      @subscriber.on_failure error(0, e.to_s)
      close nil
      return
    end
    
    @bundles[bundle]=true
    bundle.send_GET @last_modified, @etag
    async.listen bundle
  end
end

#setup_bundle(b) ⇒ Object 



1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
 
# File 'lib/nchan_tools/pubsub.rb', line 1069

def setup_bundle(b)
  b.buffer_body!
  b.on_response do |code, headers, body|
    @subscriber.waiting-=1
    # Headers and body is all parsed
    b.last_modified = headers["Last-Modified"]
    b.etag = headers["Etag"]
    b.request_time = Time.now.to_f - b.time_requested
    if request_code_ok(code, b)
      on_message_ret=nil
      Message.each_multipart_message(headers["Content-Type"], body) do |content_type, msg_body, multi|
        unless @nomsg
          msg=Message.new msg_body.dup
          msg.content_type=content_type
          unless multi
            msg.last_modified= headers["Last-Modified"]
            msg.etag= headers["Etag"]
          end
        else
          msg=msg_body.dup
        end
        
        on_message_ret= @subscriber.on_message(msg, b)
      end
      
      unless on_message_ret == false
        @subscriber.waiting+=1
        b.send_GET
      else
        @subscriber.finished+=1
        close b
      end
    end
  end
  
  b.on_error do |msg, err|
    handle_bundle_error b, msg, err
  end
end

#stop(msg = "Stopped", src_bundle = nil) ⇒ Object 



997
998
999
1000
1001
1002
1003
 
# File 'lib/nchan_tools/pubsub.rb', line 997

def stop(msg="Stopped", src_bundle=nil)
  super msg, (@bundles.first && @bundles.first.first)
  @bundles.each do |b, v|
    close b
  end
  @timer.cancel if @timer
end