Class: Deepstream::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/deepstream.rb

Instance Method Summary collapse

Constructor Details

#initialize(address, port = 6021) ⇒ Client

Returns a new instance of Client.



58
59
60
 
# File 'lib/deepstream.rb', line 58

def initialize(address, port = 6021)
  @address, @port, @unread_msg, @event_callbacks, @records = address, port, nil, {}, {}
end

Instance Method Details

#_connect ⇒ Object 



89
90
91
92
93
 
# File 'lib/deepstream.rb', line 89

def _connect
  _open_socket
  @connected = true
  @connected = _write_and_read(%w{A REQ {}}) { |msg| msg == %w{A A} }
end

#_fire_event_callback(msg) ⇒ Object 



123
124
125
 
# File 'lib/deepstream.rb', line 123

def _fire_event_callback(msg)
  @event_callbacks[msg[2]].tap { |cb| Thread.start { cb.(_parse_data(msg[3])) } if cb }
end

#_open_socket ⇒ Object 



79
80
81
82
83
84
85
86
87
 
# File 'lib/deepstream.rb', line 79

def _open_socket
  Timeout.timeout(2) { @socket = TCPSocket.new(@address, @port) }
  Thread.start do
    loop { _process_msg(@socket.gets(30.chr).tap { |m| break m.chomp(30.chr).split(31.chr) if m }) }
  end
rescue
  print Time.now.to_s[/.+ .+ /], "Can't connect to deepstream server\n"
  raise
end

#_parse_data(payload) ⇒ Object 



138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/deepstream.rb', line 138

def _parse_data(payload)
  case payload[0]
  when 'O' then JSON.parse(payload[1..-1], object_class: OpenStruct)
  when '{' then JSON.parse(payload, object_class: OpenStruct)
  when 'S' then payload[1..-1]
  when 'N' then payload[1..-1].to_f
  when 'T' then true
  when 'F' then false
  when 'L' then nil
  end
end

#_process_msg(msg) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
 
# File 'lib/deepstream.rb', line 108

def _process_msg(msg)
  case msg[0..1]
  when %w{E EVT} then _fire_event_callback(msg)
  when %w{R P} then @records[msg[2]]._patch(msg[3], msg[4], _parse_data(msg[5]))
  when %w{R U} then @records[msg[2]]._update(msg[3], _parse_data(msg[4]))
  when %w{R A} then @records.delete(msg[3]) if msg[2] == 'D'
  when %w{E A} then nil
  else @unread_msg = msg
  end
end

#_read ⇒ Object 



119
120
121
 
# File 'lib/deepstream.rb', line 119

def _read
  loop { break @unread_msg || (next sleep 0.01) }.tap { @unread_msg = nil }
end

#_typed(value) ⇒ Object 



127
128
129
130
131
132
133
134
135
136
 
# File 'lib/deepstream.rb', line 127

def _typed(value)
  case value
  when Hash then "O#{value.to_json}"
  when String then "S#{value}"
  when Numeric then "N#{value}"
  when TrueClass then 'T'
  when FalseClass then 'F'
  when NilClass then 'L'
  end
end

#_write(*args) ⇒ Object 



101
102
103
104
105
106
 
# File 'lib/deepstream.rb', line 101

def _write(*args)
  _connect unless @connected
  @socket.write(args.join(31.chr) + 30.chr)
rescue
  @connected = false
end

#_write_and_read(*args) {|_read| ... } ⇒ Object

Yields:



95
96
97
98
99
 
# File 'lib/deepstream.rb', line 95

def _write_and_read(*args)
  @unread_msg = nil
  _write(*args)
  yield _read if block_given?
end

#emit(event, value = nil) ⇒ Object 



62
63
64
 
# File 'lib/deepstream.rb', line 62

def emit(event, value = nil)
  _write('E', 'EVT', event, _typed(value))
end

#get(record_name) ⇒ Object 



71
72
73
74
75
76
77
 
# File 'lib/deepstream.rb', line 71

def get(record_name)
  @records[record_name] ||= (
    _write_and_read('R', 'CR', record_name)
    msg = _read
    Deepstream::Record.new(self, record_name, _parse_data(msg[4]), msg[3].to_i)
  )
end

#on(event, &block) ⇒ Object 



66
67
68
69
 
# File 'lib/deepstream.rb', line 66

def on(event, &block)
  _write_and_read('E', 'S', event)
  @event_callbacks[event] = block
end