Class: Combi::WebSocket

Inherits:
Bus
  • Object
show all
Defined in:
lib/combi/buses/web_socket.rb

Defined Under Namespace

Classes: Client, Server

Constant Summary

Constants inherited from Bus

Bus::RPC_DEFAULT_TIMEOUT

Instance Attribute Summary collapse

Attributes inherited from Bus

#routes

Instance Method Summary collapse

Methods inherited from Bus

#add_service, #enable, #restart!

Constructor Details

#initialize(options) ⇒ WebSocket

Returns a new instance of WebSocket.



95
96
97
 
# File 'lib/combi/buses/web_socket.rb', line 95

def initialize(options)
  super
end

Instance Attribute Details

#ready ⇒ Object (readonly)

Returns the value of attribute ready.



93
94
95
 
# File 'lib/combi/buses/web_socket.rb', line 93

def ready
  @ready
end

Instance Method Details

#manage_request(env, handler) ⇒ Object 



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/combi/buses/web_socket.rb', line 118

def manage_request(env, handler)
  require 'faye/websocket'

  return unless Faye::WebSocket.websocket?(env)
  @ws = ws = Faye::WebSocket.new(env)
  session = nil

  ws.on :message do |event|
    @machine.on_message(ws, session, event.data)
  end

  ws.on :open do |event|
    session = @machine.on_open(ws, handler)
  end

  ws.on :close do |event|
    @machine.on_close(session)
  end
  # Return async Rack response
  ws.rack_response
end

#manage_ws_event(ws, handler) ⇒ Object 



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/combi/buses/web_socket.rb', line 140

def manage_ws_event(ws, handler)
  session = nil

  ws.onmessage do |raw_message|
    @machine.on_message(ws, session, raw_message)
  end

  ws.onopen do |handshake|
    session = @machine.on_open(ws, handler)
  end

  ws.onclose do
    @machine.on_close(session)
  end
end

#on_message(ws, message, session = nil) ⇒ Object 



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/combi/buses/web_socket.rb', line 156

def on_message(ws, message, session = nil)
  if message[:correlation_id] && message.has_key?(:result)
    @response_store.handle_rpc_response(message)
    Combi.logger.debug {"Handled response with correlation_id #{message[:correlation_id]} - #{message.inspect[0..500]}"}
    return
  end
  service_name = message[:service]
  kind = message[:kind]
  payload = message[:payload] || {}
  payload[:session] = session
  begin
    response = invoke_service(service_name, kind, payload)
  rescue RuntimeError => e
    response = {error: {klass: e.class.name, message: e.message, backtrace: e.backtrace } }
  end

  if message[:correlation_id]
    # The client is insterested in a response
    msg = {result: 'ok', correlation_id: message[:correlation_id]}

    response.callback do |service_response|
      msg[:response] = service_response
      send_response ws, msg
    end
    response.errback do |service_response|
      msg[:response] = {error: service_response}
      send_response ws, msg
    end
  end
end

#post_initialize ⇒ Object 



99
100
101
102
103
104
105
106
107
108
 
# File 'lib/combi/buses/web_socket.rb', line 99

def post_initialize
  @ready = EventMachine::DefaultDeferrable.new
  @response_store = Combi::ResponseStore.new
  if @options[:remote_api]
    require 'eventmachine'
    @machine = Client.new(@options[:remote_api], @options[:handler], self)
  else
    @machine = Server.new(self)
  end
end

#request(name, kind, message, timeout: RPC_DEFAULT_TIMEOUT, fast: false, ws: nil) ⇒ Object 



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/combi/buses/web_socket.rb', line 192

def request(name, kind, message, timeout: RPC_DEFAULT_TIMEOUT, fast: false, ws: nil)
  msg = {
    service: name,
    kind: kind,
    payload: message
  }
  if fast
    waiter = nil
  else
    correlation_id = Combi::Correlation.generate
    msg[:correlation_id] = correlation_id
    waiter = @response_store.wait_for(correlation_id, timeout)
  end
  @ready.callback do |r|
    web_socket = @machine.respond_to?(:ws) ? @machine.ws : ws
    unless web_socket.nil?
      serialized = Yajl::Encoder.encode msg
      Combi.logger.debug {"sending request #{serialized}"}
      web_socket.send serialized
    end
  end
  waiter
end

#send_response(ws, message) ⇒ Object 



187
188
189
190
 
# File 'lib/combi/buses/web_socket.rb', line 187

def send_response(ws, message)
  serialized = Yajl::Encoder.encode message
  ws.send serialized
end

#start! ⇒ Object 



110
111
112
 
# File 'lib/combi/buses/web_socket.rb', line 110

def start!
  @machine.start!
end

#stop! ⇒ Object 



114
115
116
 
# File 'lib/combi/buses/web_socket.rb', line 114

def stop!
  @machine.stop!
end