Class: RestBuilder::EventSource

Inherits:
Struct
  • Object
show all
Defined in:
lib/rest-builder/event_source.rb

Constant Summary collapse

READ_WAIT = 
35

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#client ⇒ Object

Returns the value of attribute client

Returns:

  • (Object) —

    the current value of client



6
7
8
 
# File 'lib/rest-builder/event_source.rb', line 6

def client
  @client
end

#opts ⇒ Object

Returns the value of attribute opts

Returns:

  • (Object) —

    the current value of opts



6
7
8
 
# File 'lib/rest-builder/event_source.rb', line 6

def opts
  @opts
end

#path ⇒ Object

Returns the value of attribute path

Returns:

  • (Object) —

    the current value of path



6
7
8
 
# File 'lib/rest-builder/event_source.rb', line 6

def path
  @path
end

#query ⇒ Object

Returns the value of attribute query

Returns:

  • (Object) —

    the current value of query



6
7
8
 
# File 'lib/rest-builder/event_source.rb', line 6

def query
  @query
end

#socket ⇒ Object

Returns the value of attribute socket

Returns:

  • (Object) —

    the current value of socket



6
7
8
 
# File 'lib/rest-builder/event_source.rb', line 6

def socket
  @socket
end

Instance Method Details

#close ⇒ Object 



25
26
27
28
 
# File 'lib/rest-builder/event_source.rb', line 25

def close
  socket && socket.close
rescue IOError
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


21
22
23
 
# File 'lib/rest-builder/event_source.rb', line 21

def closed?
  !!(socket && socket.closed?) || @closed
end

#onerror(error = nil, sock = nil, &cb) ⇒ Object

would also be called upon closing, would always be called at least once



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/rest-builder/event_source.rb', line 64

def onerror error=nil, sock=nil, &cb
  if block_given?
    @onerror = cb
  else
    begin
      Promise.set_backtrace(error)
      @onerror.call(error, sock) if @onerror
      onreconnect(error, sock)
    rescue Exception
      mutex.synchronize do
        @closed = true
        condv.signal # so we can't be reconnecting, need to try to unblock
      end
      raise
    end
  end
  self
end

#onmessage(event = nil, data = nil, sock = nil, &cb) ⇒ Object 



54
55
56
57
58
59
60
61
 
# File 'lib/rest-builder/event_source.rb', line 54

def onmessage event=nil, data=nil, sock=nil, &cb
  if block_given?
    @onmessage = cb
  elsif @onmessage
    @onmessage.call(event, data, sock)
  end
  self
end

#onopen(sock = nil, &cb) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/rest-builder/event_source.rb', line 36

def onopen sock=nil, &cb
  if block_given?
    @onopen = cb
  else
    self.socket = sock # for you to track the socket
    @onopen.call(sock) if @onopen
    onmessage_for(sock)
  end
  self
rescue Exception => e
  begin # close the socket since we're going to stop anyway
    sock.close # if we don't close it, client might wait forever
  rescue IOError
  end
  # let the client has a chance to handle this, and make signal
  onerror(e, sock)
end

#onreconnect(error = nil, sock = nil, &cb) ⇒ Object

would be called upon closing, and would try to reconnect if a callback is set and return true



85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/rest-builder/event_source.rb', line 85

def onreconnect error=nil, sock=nil, &cb
  if block_given?
    @onreconnect = cb
  elsif closed? && @onreconnect && @onreconnect.call(error, sock)
    reconnect
  else
    mutex.synchronize do
      @closed = true
      condv.signal # we could be closing, let's try to unblock it
    end
  end
  self
end

#start ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/rest-builder/event_source.rb', line 9

def start
  self.mutex = Mutex.new
  self.condv = ConditionVariable.new
  @onopen      ||= nil
  @onmessage   ||= nil
  @onerror     ||= nil
  @onreconnect ||= nil
  @closed      ||= false
  reconnect
  self
end

#wait ⇒ Object

Raises:



30
31
32
33
34
 
# File 'lib/rest-builder/event_source.rb', line 30

def wait
  raise Error.new("Not yet started for: #{self}") unless mutex
  mutex.synchronize{ condv.wait(mutex) until closed? } unless closed?
  self
end