Class: NchanTools::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.



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

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.



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

def timeout
  @timeout
end

Class Method Details

.aliases ⇒ Object 



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

def self.aliases
  [:longpoll]
end

Instance Method Details

#close(bundle) ⇒ Object 



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

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 



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

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

#listen(bundle) ⇒ Object 



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

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 



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

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 



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

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 



1004
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
 
# File 'lib/nchan_tools/pubsub.rb', line 1004

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 



1068
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
 
# File 'lib/nchan_tools/pubsub.rb', line 1068

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 



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

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