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, #log, #restart!

Constructor Details

#initialize(options) ⇒ WebSocket

Returns a new instance of WebSocket.



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

def initialize(options)
  super
end

Instance Attribute Details

#ready ⇒ Object (readonly)

Returns the value of attribute ready.



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

def ready
  @ready
end

Instance Method Details

#manage_request(env, handler) ⇒ Object 



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

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 



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

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 



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
186
 
# File 'lib/combi/buses/web_socket.rb', line 157

def on_message(ws, message, session = nil)
  if message[:correlation_id] && message.has_key?(:result)
    @response_store.handle_rpc_response(message)
    log "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 



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

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 



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

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
      log "sending request #{serialized}"
      web_socket.send serialized
    end
  end
  waiter
end

#send_response(ws, message) ⇒ Object 



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

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

#start! ⇒ Object 



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

def start!
  @machine.start!
end

#stop! ⇒ Object 



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

def stop!
  @machine.stop!
end