* p2p initial version implemented and particially tested
This commit is contained in:
parent
9a055fd2c2
commit
c711fa932b
|
|
@ -2,6 +2,7 @@ require "logging"
|
|||
require "jam_ruby"
|
||||
require "jam_websockets/version"
|
||||
require "jam_websockets/session_error"
|
||||
require "jam_websockets/permission_error"
|
||||
require "jam_websockets/client_context"
|
||||
require "jam_websockets/message"
|
||||
require "jam_websockets/router"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,4 @@
|
|||
class PermissionError < Exception
|
||||
|
||||
end
|
||||
|
||||
|
|
@ -11,37 +11,40 @@ include Jampb
|
|||
|
||||
# add new field to client connection
|
||||
module EventMachine
|
||||
module WebSocket
|
||||
class Connection < EventMachine::Connection
|
||||
module WebSocket
|
||||
class Connection < EventMachine::Connection
|
||||
attr_accessor :encode_json, :client_id # client_id is uuid we give to each client to track them as we like
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
module JamWebsockets
|
||||
|
||||
|
||||
class Router
|
||||
|
||||
attr_accessor :user_context_lookup, :session_context_lookup
|
||||
|
||||
attr_accessor :user_context_lookup, :session_context_lookup
|
||||
|
||||
def initialize(options={})
|
||||
@log = Logging.logger[self]
|
||||
@pending_clients = Set.new # clients that have connected to server, but not logged in.
|
||||
@clients = {} # clients that have logged in
|
||||
@user_context_lookup = {} # lookup a set of client_contexts by user_id
|
||||
@session_context_lookup = {} # lookup a set of client_contexts by session_id
|
||||
@user_context_lookup = {} # lookup a set of client_contexts by user_id
|
||||
@session_context_lookup = {} # lookup a set of client_contexts by session_id
|
||||
@client_lookup = {} # lookup a client by client_id
|
||||
@sessions_exchange = nil
|
||||
@connection = nil
|
||||
@channel = nil
|
||||
@users_exchange = nil
|
||||
@message_factory = JamRuby::MessageFactory.new
|
||||
@semaphore = Mutex.new
|
||||
@user_topic = nil
|
||||
@user_subscription = nil
|
||||
@session_topic = nil
|
||||
@session_subscription = nil
|
||||
@thread_pool = nil
|
||||
@semaphore = Mutex.new
|
||||
@user_topic = nil
|
||||
@user_subscription = nil
|
||||
@session_topic = nil
|
||||
@session_subscription = nil
|
||||
@client_topic = nil
|
||||
@client_subscription = nil
|
||||
@thread_pool = nil
|
||||
end
|
||||
|
||||
def start(options = {})
|
||||
|
|
@ -49,7 +52,7 @@ module JamWebsockets
|
|||
@log.info "startup"
|
||||
|
||||
begin
|
||||
@thread_pool = Executors.new_fixed_thread_pool(8)
|
||||
@thread_pool = Executors.new_fixed_thread_pool(8)
|
||||
@connection = HotBunnies.connect(:host => options[:host], :port => options[:port])
|
||||
@channel = @connection.create_channel
|
||||
@channel.prefetch = 10
|
||||
|
|
@ -62,119 +65,143 @@ module JamWebsockets
|
|||
|
||||
end
|
||||
|
||||
def add_user(context)
|
||||
user_contexts = @user_context_lookup[context.user.id]
|
||||
if user_contexts.nil?
|
||||
user_contexts = Set.new
|
||||
@user_context_lookup[context.user.id] = user_contexts
|
||||
end
|
||||
def add_client(client_id, client)
|
||||
|
||||
user_contexts.add(context)
|
||||
end
|
||||
# should never occur
|
||||
if @client_lookup.has_key?(client_id)
|
||||
@log.warn "client_id #{client_id} connected while the old connection has not yet been terminated"
|
||||
end
|
||||
|
||||
def remove_user(context)
|
||||
user_contexts = @user_context_lookup[context.user.id]
|
||||
if user_contexts.nil?
|
||||
@log.warn "user can not be removed #{context}"
|
||||
else
|
||||
# delete the context from set of user contexts
|
||||
user_contexts.delete(context)
|
||||
@client_lookup[client_id] = client
|
||||
end
|
||||
|
||||
# if last user context, delete entire set (memory leak concern)
|
||||
if user_contexts.length == 0
|
||||
@user_context_lookup.delete(context.user.id)
|
||||
end
|
||||
end
|
||||
end
|
||||
def remove_client(client_id, client)
|
||||
deleted = @client_lookup.delete(client_id)
|
||||
|
||||
def add_session(context)
|
||||
session_contexts = @session_context_lookup[context.session.id]
|
||||
if session_contexts.nil?
|
||||
session_contexts = Set.new
|
||||
@session_context_lookup[context.session.id] = session_contexts
|
||||
end
|
||||
if deleted.nil?
|
||||
@log.warn "unable to delete #{client_id} from client_lookup"
|
||||
elsif deleted == client
|
||||
# put it back--this is only possible if add_client hit the 'old connection' path
|
||||
# so in other words if this happens:
|
||||
# add_client(1, clientX)
|
||||
# add_client(1, clientY) # but clientX is essentially defunct - this could happen due to a bug in client, or EM doesn't notify always of connection close in time
|
||||
# remove_client(1, clientX) -- this check maintains that clientY stays as the current client in the hash
|
||||
@client_lookup[client_id] = client
|
||||
end
|
||||
end
|
||||
|
||||
session_contexts.add(context)
|
||||
end
|
||||
def add_user(context)
|
||||
user_contexts = @user_context_lookup[context.user.id]
|
||||
if user_contexts.nil?
|
||||
user_contexts = Set.new
|
||||
@user_context_lookup[context.user.id] = user_contexts
|
||||
end
|
||||
|
||||
def remove_session(context)
|
||||
session_contexts = @session_context_lookup[context.session.id]
|
||||
if session_contexts.nil?
|
||||
@log.warn "session can not be removed #{context}"
|
||||
else
|
||||
# delete the context from set of session contexts
|
||||
session_contexts.delete(context)
|
||||
user_contexts.add(context)
|
||||
end
|
||||
|
||||
# if last session context, delete entire set (memory leak concern)
|
||||
if session_contexts.length == 0
|
||||
@session_context_lookup.delete(context.session.id)
|
||||
end
|
||||
def remove_user(context)
|
||||
user_contexts = @user_context_lookup[context.user.id]
|
||||
if user_contexts.nil?
|
||||
@log.warn "user can not be removed #{context}"
|
||||
else
|
||||
# delete the context from set of user contexts
|
||||
user_contexts.delete(context)
|
||||
|
||||
context.session = nil
|
||||
end
|
||||
end
|
||||
# if last user context, delete entire set (memory leak concern)
|
||||
if user_contexts.length == 0
|
||||
@user_context_lookup.delete(context.user.id)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def add_session(context)
|
||||
session_contexts = @session_context_lookup[context.session.id]
|
||||
if session_contexts.nil?
|
||||
session_contexts = Set.new
|
||||
@session_context_lookup[context.session.id] = session_contexts
|
||||
end
|
||||
|
||||
# register topic for user messages and session messages
|
||||
def register_topics
|
||||
@users_exchange = @channel.exchange('users', :type => :topic)
|
||||
session_contexts.add(context)
|
||||
end
|
||||
|
||||
def remove_session(context)
|
||||
session_contexts = @session_context_lookup[context.session.id]
|
||||
if session_contexts.nil?
|
||||
@log.warn "session can not be removed #{context}"
|
||||
else
|
||||
# delete the context from set of session contexts
|
||||
session_contexts.delete(context)
|
||||
|
||||
# if last session context, delete entire set (memory leak concern)
|
||||
if session_contexts.length == 0
|
||||
@session_context_lookup.delete(context.session.id)
|
||||
end
|
||||
|
||||
context.session = nil
|
||||
end
|
||||
end
|
||||
|
||||
# register topic for user messages and session messages
|
||||
def register_topics
|
||||
@users_exchange = @channel.exchange('users', :type => :topic)
|
||||
@sessions_exchange = @channel.exchange('sessions', :type => :topic)
|
||||
|
||||
# create user messaging topic
|
||||
@clients_exchange = @channel.exchange('clients', :type => :topic)
|
||||
|
||||
# create user messaging topic
|
||||
@user_topic = @channel.queue("", :auto_delete => true)
|
||||
@user_topic.bind(@users_exchange, :routing_key => "user.#")
|
||||
@user_topic.purge
|
||||
|
||||
# TODO: alert friends
|
||||
|
||||
# subscribe for any messages to users
|
||||
|
||||
#@user_subscription = @user_topic.subscribe(:ack => false, :blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
@user_subscription = @user_topic.subscribe(:ack => false)
|
||||
@user_subscription.each(:blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
begin
|
||||
routing_key = headers.envelope.routing_key
|
||||
user_id = routing_key["user.".length..-1]
|
||||
@sempahore.synchronize do
|
||||
contexts = @user_context_lookup[user_id]
|
||||
# subscribe for any messages to users
|
||||
|
||||
@user_subscription = @user_topic.subscribe(:ack => false)
|
||||
@user_subscription.each(:blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
begin
|
||||
routing_key = headers.envelope.routing_key
|
||||
user_id = routing_key["user.".length..-1]
|
||||
@sempahore.synchronize do
|
||||
contexts = @user_context_lookup[user_id]
|
||||
|
||||
unless contexts.nil?
|
||||
|
||||
@log.debug "received user-directed message for session: #{user_id}"
|
||||
|
||||
unless contexts.nil?
|
||||
|
||||
@log.debug "received user-directed message for session: #{user_id}"
|
||||
|
||||
msg = Jampb::ClientMessage.parse(msg)
|
||||
contexts.each do |context|
|
||||
EM.schedule do
|
||||
@log.debug "sending user message to #{context}"
|
||||
send_to_client(context.client, msg)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
contexts.each do |context|
|
||||
EM.schedule do
|
||||
@log.debug "sending user message to #{context}"
|
||||
send_to_client(context.client, msg)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
rescue => e
|
||||
@log.error "unhandled error in messaging to client"
|
||||
end
|
||||
end
|
||||
rescue => e
|
||||
@log.error "unhandled error in messaging to client"
|
||||
@log.error e
|
||||
end
|
||||
end
|
||||
|
||||
@session_topic = @channel.queue("", :auto_delete => true)
|
||||
@session_topic = @channel.queue("", :auto_delete => true)
|
||||
@session_topic.bind(@sessions_exchange, :routing_key => "session.#")
|
||||
@session_topic.purge
|
||||
|
||||
# subscribe for any messages to session
|
||||
#@session_subscription = @session_topic.subscribe(:ack => false, :blocking => false) do |headers, msg|
|
||||
@session_subscription = @session_topic.subscribe(:ack => false)
|
||||
@session_subscription.each(:blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
begin
|
||||
routing_key = headers.envelope.routing_key
|
||||
session_id = routing_key["session.".length..-1]
|
||||
@semaphore.synchronize do
|
||||
contexts = @session_context_lookup[session_id]
|
||||
@session_subscription.each(:blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
begin
|
||||
routing_key = headers.envelope.routing_key
|
||||
session_id = routing_key["session.".length..-1]
|
||||
@semaphore.synchronize do
|
||||
contexts = @session_context_lookup[session_id]
|
||||
|
||||
unless contexts.nil?
|
||||
|
||||
@log.debug "received session-directed message for session: #{session_id}"
|
||||
|
||||
unless contexts.nil?
|
||||
|
||||
@log.debug "received session-directed message for session: #{session_id}"
|
||||
|
||||
msg = Jampb::ClientMessage.parse(msg)
|
||||
|
||||
# ok, its very odd to have your own message that you sent bounce back to you.
|
||||
|
|
@ -183,94 +210,132 @@ module JamWebsockets
|
|||
properties = headers.properties unless headers.nil?
|
||||
inner_headers = properties.headers unless properties.nil?
|
||||
origin_client_id = inner_headers["client_id"]
|
||||
|
||||
|
||||
# counter-intuitively, even though a string is passed in when you send the header, an (apparently) auto-generated class is sent back which, if you to_s, returns the original value
|
||||
origin_client_id = origin_client_id.to_s unless origin_client_id.nil?
|
||||
|
||||
|
||||
@log.debug "message received from client #{origin_client_id}"
|
||||
contexts.each do |context|
|
||||
contexts.each do |context|
|
||||
if context.client.client_id != origin_client_id
|
||||
EM.schedule do
|
||||
@log.debug "sending session message to #{context}"
|
||||
send_to_client(context.client, msg)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
rescue => e
|
||||
@log.error "unhandled error in messaging to client"
|
||||
end
|
||||
rescue => e
|
||||
@log.error "unhandled error in messaging to client"
|
||||
@log.error e
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def send_to_client(client, msg)
|
||||
@client_topic = @channel.queue("", :auto_delete => true)
|
||||
@client_topic.bind(@clients_exchange, :routing_key => "client.#")
|
||||
@client_topic.purge
|
||||
|
||||
# subscribe for any p2p messages to a client
|
||||
@client_subscription = @client_topic.subscribe(:ack => false)
|
||||
@client_subscription.each(:blocking => false, :executor => @threadpool) do |headers, msg|
|
||||
begin
|
||||
routing_key = headers.envelope.routing_key
|
||||
client_id = routing_key["client.".length..-1]
|
||||
@semaphore.synchronize do
|
||||
client = @client_lookup[client_id]
|
||||
|
||||
unless client.nil?
|
||||
msg = Jampb::ClientMessage.parse(msg)
|
||||
|
||||
properties = headers.properties unless headers.nil?
|
||||
inner_headers = properties.headers unless properties.nil?
|
||||
origin_client_id = inner_headers["client_id"]
|
||||
|
||||
@log.debug "p2p message received from client #{origin_client_id} to client #{client_id}"
|
||||
EM.schedule do
|
||||
@log.debug "sending p2p message to #{client_id}"
|
||||
send_to_client(client, msg)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
rescue => e
|
||||
@log.error "unhandled error in messaging to client"
|
||||
@log.error e
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def send_to_client(client, msg)
|
||||
if client.encode_json
|
||||
client.send(msg.to_json.to_s)
|
||||
else
|
||||
# this is so odd that this is necessary from an API perspective. but searching through the source code... it's all I could find in em-websocket for allowing a binary message to be sent
|
||||
client.instance_variable_get(:@handler).send_frame(:binary, msg.to_s)
|
||||
# this is so odd that this is necessary from an API perspective. but searching through the source code... it's all I could find in em-websocket for allowing a binary message to be sent
|
||||
client.instance_variable_get(:@handler).send_frame(:binary, msg.to_s)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def cleanup()
|
||||
# shutdown topic listeners and mq connection
|
||||
begin
|
||||
if !@user_subscription.nil? && @user_subscription.active?
|
||||
@log.debug "cleaning up user subscription"
|
||||
@user_subscription.cancel
|
||||
@user_subscription.shutdown!
|
||||
end
|
||||
# shutdown topic listeners and mq connection
|
||||
begin
|
||||
if !@user_subscription.nil? && @user_subscription.active?
|
||||
@log.debug "cleaning up user subscription"
|
||||
@user_subscription.cancel
|
||||
@user_subscription.shutdown!
|
||||
end
|
||||
|
||||
if !@session_subscription.nil? && @session_subscription.active?
|
||||
@log.debug "cleaning up session subscription"
|
||||
@session_subscription.cancel
|
||||
@session_subscription.shutdown!
|
||||
end
|
||||
if !@session_subscription.nil? && @session_subscription.active?
|
||||
@log.debug "cleaning up session subscription"
|
||||
@session_subscription.cancel
|
||||
@session_subscription.shutdown!
|
||||
end
|
||||
|
||||
rescue => e
|
||||
@log.debug "unable to cancel subscription on cleanup: #{e}"
|
||||
end
|
||||
if !@client_subscription.nil? && @client_subscription.active?
|
||||
@log.debug "cleaning up client subscription"
|
||||
@client_subscription.cancel
|
||||
@client_subscription.shutdown!
|
||||
end
|
||||
|
||||
@thread_pool.shutdown
|
||||
rescue => e
|
||||
@log.debug "unable to cancel subscription on cleanup: #{e}"
|
||||
end
|
||||
|
||||
if !@channel.nil?
|
||||
@channel.close
|
||||
end
|
||||
@thread_pool.shutdown
|
||||
|
||||
if !@connection.nil?
|
||||
@connection.close
|
||||
end
|
||||
if !@channel.nil?
|
||||
@channel.close
|
||||
end
|
||||
|
||||
# tear down each individual client
|
||||
@clients.each do |client, context|
|
||||
cleanup_client(client)
|
||||
end
|
||||
end
|
||||
if !@connection.nil?
|
||||
@connection.close
|
||||
end
|
||||
|
||||
def stop
|
||||
@log.info "shutdown"
|
||||
cleanup
|
||||
end
|
||||
# tear down each individual client
|
||||
@clients.each do |client, context|
|
||||
cleanup_client(client)
|
||||
end
|
||||
end
|
||||
|
||||
def new_client(client)
|
||||
def stop
|
||||
@log.info "shutdown"
|
||||
cleanup
|
||||
end
|
||||
|
||||
# give a unique ID to this client. This is used to prevent session messages
|
||||
# from echoing back to the sender, for instance.
|
||||
client.client_id = UUIDTools::UUID.random_create.to_s
|
||||
def new_client(client)
|
||||
|
||||
@semaphore.synchronize do
|
||||
@pending_clients.add(client)
|
||||
end
|
||||
@semaphore.synchronize do
|
||||
@pending_clients.add(client)
|
||||
end
|
||||
|
||||
# default to using json instead of pb
|
||||
client.encode_json = true
|
||||
client.encode_json = true
|
||||
|
||||
client.onopen {
|
||||
#binding.pry
|
||||
@log.debug "client connected #{client}"
|
||||
|
||||
client.onopen {
|
||||
#binding.pry
|
||||
@log.debug "client connected #{client}"
|
||||
|
||||
# check for '?pb' or '?pb=true' in url query parameters
|
||||
query_pb = client.request["query"]["pb"]
|
||||
|
||||
|
|
@ -280,10 +345,10 @@ module JamWebsockets
|
|||
|
||||
}
|
||||
|
||||
client.onclose {
|
||||
@log.debug "Connection closed"
|
||||
client.onclose {
|
||||
@log.debug "Connection closed"
|
||||
|
||||
cleanup_client(client)
|
||||
cleanup_client(client)
|
||||
}
|
||||
|
||||
client.onerror { |error|
|
||||
|
|
@ -294,7 +359,7 @@ module JamWebsockets
|
|||
end
|
||||
|
||||
cleanup_client(client)
|
||||
client.close_websocket
|
||||
client.close_websocket
|
||||
}
|
||||
|
||||
client.onmessage { |msg|
|
||||
|
|
@ -303,79 +368,90 @@ module JamWebsockets
|
|||
# TODO: set a max message size before we put it through PB?
|
||||
# TODO: rate limit?
|
||||
|
||||
|
||||
|
||||
begin
|
||||
if client.encode_json
|
||||
#example: {"type":"LOGIN", "target":"server", "login" : {"username":"hi"}}
|
||||
parse = JSON.parse(msg)
|
||||
pb_msg = Jampb::ClientMessage.json_create(parse)
|
||||
if client.encode_json
|
||||
#example: {"type":"LOGIN", "route_to":"server", "login" : {"username":"hi"}}
|
||||
parse = JSON.parse(msg)
|
||||
pb_msg = Jampb::ClientMessage.json_create(parse)
|
||||
self.route(pb_msg, client)
|
||||
else
|
||||
pb_msg = Jampb::ClientMessage.parse(msg.to_s)
|
||||
self.route(pb_msg, client)
|
||||
end
|
||||
rescue SessionError => e
|
||||
@log.info "ending client session deliberately due to malformed client behavior. reason=#{e}"
|
||||
begin
|
||||
else
|
||||
pb_msg = Jampb::ClientMessage.parse(msg.to_s)
|
||||
self.route(pb_msg, client)
|
||||
end
|
||||
rescue SessionError => e
|
||||
@log.info "ending client session deliberately due to malformed client behavior. reason=#{e}"
|
||||
begin
|
||||
# wrap the message up and send it down
|
||||
error_msg = @message_factory.server_rejection_error(e.to_s)
|
||||
send_to_client(client, error_msg)
|
||||
send_to_client(client, error_msg)
|
||||
ensure
|
||||
client.close_websocket
|
||||
client.close_websocket
|
||||
cleanup_client(client)
|
||||
end
|
||||
rescue PermissionError => e
|
||||
@log.info "permission error. reason=#{e.to_s}"
|
||||
@log.info e
|
||||
|
||||
# wrap the message up and send it down
|
||||
error_msg = @message_factory.server_permission_error(msg.message_id, e.to_s)
|
||||
send_to_client(client, error_msg)
|
||||
rescue => e
|
||||
@log.error "ending client session due to server programming or runtime error. reason=#{e.to_s}"
|
||||
@log.error e
|
||||
|
||||
begin
|
||||
@log.error e
|
||||
|
||||
begin
|
||||
# wrap the message up and send it down
|
||||
error_msg = @message_factory.server_generic_error(e.to_s)
|
||||
send_to_client(client, error_msg)
|
||||
send_to_client(client, error_msg)
|
||||
ensure
|
||||
client.close_websocket
|
||||
client.close_websocket
|
||||
cleanup_client(client)
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
}
|
||||
|
||||
end
|
||||
|
||||
# removes all resources associated with a client
|
||||
# removes all resources associated with a client
|
||||
def cleanup_client(client)
|
||||
|
||||
@semaphore.synchronize do
|
||||
pending = @pending_clients.delete?(client)
|
||||
@semaphore.synchronize do
|
||||
pending = @pending_clients.delete?(client)
|
||||
|
||||
if !pending.nil?
|
||||
@log.debug "cleaning up pending client #{client}"
|
||||
else
|
||||
context = @clients.delete(client)
|
||||
if !pending.nil?
|
||||
@log.debug "cleaning up pending client #{client}"
|
||||
else
|
||||
|
||||
if !context.nil?
|
||||
|
||||
remove_user(context)
|
||||
remove_client(client.client_id, client)
|
||||
|
||||
if !context.session.nil?
|
||||
remove_session(context)
|
||||
end
|
||||
else
|
||||
@log.debug "skipping duplicate cleanup attempt of authorized client"
|
||||
end
|
||||
context = @clients.delete(client)
|
||||
|
||||
end
|
||||
end
|
||||
if !context.nil?
|
||||
|
||||
remove_user(context)
|
||||
|
||||
if !context.session.nil?
|
||||
remove_session(context)
|
||||
end
|
||||
else
|
||||
@log.debug "skipping duplicate cleanup attempt of authorized client"
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def route(client_msg, client)
|
||||
message_type = @message_factory.get_message_type(client_msg)
|
||||
|
||||
raise SessionError, "unknown message type received: #{client_msg.type}" if message_type.nil?
|
||||
|
||||
@log.debug("msg received #{message_type}")
|
||||
message_type = @message_factory.get_message_type(client_msg)
|
||||
|
||||
raise SessionError, 'client_msg.target is null' if client_msg.target.nil?
|
||||
raise SessionError, "unknown message type received: #{client_msg.type}" if message_type.nil?
|
||||
|
||||
@log.debug("msg received #{message_type}")
|
||||
|
||||
raise SessionError, 'client_msg.route_to is null' if client_msg.route_to.nil?
|
||||
|
||||
if @pending_clients.include? client and client_msg.type != ClientMessage::Type::LOGIN
|
||||
# this client has not logged in and is trying to send a non-login message
|
||||
|
|
@ -386,18 +462,23 @@ module JamWebsockets
|
|||
|
||||
handle_server_directed(client_msg, client)
|
||||
|
||||
elsif @message_factory.client_directed? client_msg
|
||||
|
||||
to_client_id = client_msg.route_to[MessageFactory::CLIENT_TARGET_PREFIX.length..-1]
|
||||
handle_client_directed(to_client_id, client_msg, client)
|
||||
|
||||
elsif @message_factory.session_directed? client_msg
|
||||
|
||||
session = client_msg.target[MessageFactory::SESSION_TARGET_PREFIX.length..-1]
|
||||
session = client_msg.route_to[MessageFactory::SESSION_TARGET_PREFIX.length..-1]
|
||||
handle_session_directed(session, client_msg, client)
|
||||
|
||||
elsif @message_factory.user_directed? client_msg
|
||||
|
||||
user = client_msg.target[MessageFactory::USER_PREFIX_TARGET.length..-1]
|
||||
user = client_msg.route_to[MessageFactory::USER_PREFIX_TARGET.length..-1]
|
||||
handle_user_directed(user, client_msg, client)
|
||||
|
||||
else
|
||||
raise SessionError, "client_msg.target is unknown type: #{client_msg.target}"
|
||||
raise SessionError, "client_msg.route_to is unknown type: #{client_msg.route_to}"
|
||||
end
|
||||
|
||||
end
|
||||
|
|
@ -408,9 +489,9 @@ module JamWebsockets
|
|||
|
||||
handle_login(client_msg.login, client)
|
||||
|
||||
elsif client_msg.type == ClientMessage::Type::HEARTBEAT
|
||||
elsif client_msg.type == ClientMessage::Type::HEARTBEAT
|
||||
|
||||
handle_heartbeat(client_msg.heartbeat, client)
|
||||
handle_heartbeat(client_msg.heartbeat, client)
|
||||
|
||||
elsif client_msg.type == ClientMessage::Type::LOGIN_MUSIC_SESSION
|
||||
|
||||
|
|
@ -421,7 +502,7 @@ module JamWebsockets
|
|||
handle_leave_music_session(client_msg.leave_music_session, client)
|
||||
|
||||
else
|
||||
raise SessionError, "unknown message type '#{client_msg.type}' for #{client_msg.target}-directed message"
|
||||
raise SessionError, "unknown message type '#{client_msg.type}' for #{client_msg.route_to}-directed message"
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -430,36 +511,47 @@ module JamWebsockets
|
|||
username = login.username if login.value_for_tag(1)
|
||||
password = login.password if login.value_for_tag(2)
|
||||
token = login.token if login.value_for_tag(3)
|
||||
client_id = login.client_id if login.value_for_tag(4)
|
||||
|
||||
user = valid_login(username, password, token)
|
||||
# you don't have to supply client_id in login--if you don't, we'll generate one
|
||||
if client_id.nil?
|
||||
# give a unique ID to this client. This is used to prevent session messages
|
||||
# from echoing back to the sender, for instance.
|
||||
client_id = UUIDTools::UUID.random_create.to_s
|
||||
end
|
||||
|
||||
client.client_id = client_id
|
||||
|
||||
user = valid_login(username, password, token, client_id)
|
||||
|
||||
if !user.nil?
|
||||
|
||||
@log.debug "user #{user.email} logged in"
|
||||
|
||||
# respond with LOGIN_ACK to let client know it was successful
|
||||
#binding.pry
|
||||
remote_port, remote_ip = Socket.unpack_sockaddr_in(client.get_peername)
|
||||
login_ack = @message_factory.login_ack(remote_ip)
|
||||
send_to_client(client, login_ack)
|
||||
#binding.pry
|
||||
remote_port, remote_ip = Socket.unpack_sockaddr_in(client.get_peername)
|
||||
login_ack = @message_factory.login_ack(remote_ip, client_id, user.remember_token)
|
||||
send_to_client(client, login_ack)
|
||||
|
||||
# remove from pending_queue
|
||||
@semaphore.synchronize do
|
||||
@pending_clients.delete(client)
|
||||
@semaphore.synchronize do
|
||||
@pending_clients.delete(client)
|
||||
|
||||
# add a tracker for this user
|
||||
context = ClientContext.new(user, client)
|
||||
@clients[client] = context
|
||||
add_user(context)
|
||||
end
|
||||
# add a tracker for this user
|
||||
context = ClientContext.new(user, client)
|
||||
@clients[client] = context
|
||||
add_user(context)
|
||||
add_client(client_id, client)
|
||||
end
|
||||
else
|
||||
raise SessionError, 'invalid login'
|
||||
end
|
||||
end
|
||||
|
||||
def handle_heartbeat(heartbeat, client)
|
||||
# todo: manage staleness
|
||||
end
|
||||
def handle_heartbeat(heartbeat, client)
|
||||
# todo: manage staleness
|
||||
end
|
||||
|
||||
def handle_join_music_session(join_music_session, client)
|
||||
# verify that the current user has the rights to actually join the music session
|
||||
|
|
@ -468,107 +560,146 @@ module JamWebsockets
|
|||
session_id = join_music_session.music_session
|
||||
|
||||
begin
|
||||
session = access_music_session?(session_id, context.user)
|
||||
@log.debug "user #{context} joining new session #{session}"
|
||||
@semaphore.synchronize do
|
||||
old_session = context.session
|
||||
if !old_session.nil?
|
||||
@log.debug "#{context} is already in session. auto-logging out to join new session."
|
||||
remove_session(context)
|
||||
end
|
||||
context.session = session
|
||||
add_session(context)
|
||||
end
|
||||
session = access_music_session(session_id, context.user)
|
||||
@log.debug "user #{context} joining new session #{session}"
|
||||
@semaphore.synchronize do
|
||||
old_session = context.session
|
||||
if !old_session.nil?
|
||||
@log.debug "#{context} is already in session. auto-logging out to join new session."
|
||||
remove_session(context)
|
||||
end
|
||||
context.session = session
|
||||
add_session(context)
|
||||
end
|
||||
rescue => e
|
||||
# send back a failure ack and bail
|
||||
@log.debug "client requested non-existent session. client:#{client.request['origin']} user:#{context.user.email}"
|
||||
@log.debug "client requested non-existent session. client:#{client.request['origin']} user:#{context.user.email}"
|
||||
login_music_session = @message_factory.login_music_session_ack(true, e.to_s)
|
||||
send_to_client(client, login_music_session)
|
||||
send_to_client(client, login_music_session)
|
||||
return
|
||||
end
|
||||
|
||||
# respond with LOGIN_MUSIC_SESSION_ACK to let client know it was successful
|
||||
login_music_session = @message_factory.login_music_session_ack(false, nil)
|
||||
send_to_client(client, login_music_session)
|
||||
|
||||
send_to_client(client, login_music_session)
|
||||
|
||||
# send 'new client' message to other members in the session
|
||||
handle_session_directed(session_id,
|
||||
@message_factory.user_joined_music_session(context.user.id, context.user.name),
|
||||
client)
|
||||
handle_session_directed(session_id,
|
||||
@message_factory.user_joined_music_session(context.user.id, context.user.name),
|
||||
client)
|
||||
end
|
||||
|
||||
def handle_leave_music_session(leave_music_session, client)
|
||||
def handle_leave_music_session(leave_music_session, client)
|
||||
|
||||
context = @clients[client]
|
||||
context = @clients[client]
|
||||
|
||||
raise SessionError, "unsupported"
|
||||
end
|
||||
raise SessionError, "unsupported"
|
||||
end
|
||||
|
||||
def valid_login(username, password, token)
|
||||
def valid_login(username, password, token, client_id)
|
||||
|
||||
if !token.nil? && token != ''
|
||||
@log.debug "logging in via token"
|
||||
# attempt login with token
|
||||
user = User.find_by_remember_token(token)
|
||||
# attempt login with token
|
||||
user = User.find_by_remember_token(token)
|
||||
|
||||
if user.nil?
|
||||
@log.debug "no user found with token"
|
||||
return false
|
||||
else
|
||||
@log.debug "#{user} login via token"
|
||||
return user
|
||||
end
|
||||
if user.nil?
|
||||
@log.debug "no user found with token"
|
||||
return false
|
||||
else
|
||||
@log.debug "#{user} login via token"
|
||||
return user
|
||||
end
|
||||
|
||||
elsif !username.nil? and !password.nil?
|
||||
elsif !username.nil? and !password.nil?
|
||||
|
||||
@log.debug "logging in via user/pass '#{username}' '#{password}'"
|
||||
# attempt login with username and password
|
||||
user = User.find_by_email(username)
|
||||
# attempt login with username and password
|
||||
user = User.find_by_email(username)
|
||||
|
||||
if !user.nil? && user.authenticate(password)
|
||||
@log.debug "#{user} login via password"
|
||||
return user
|
||||
else
|
||||
@log.debug "#{username} login failure"
|
||||
return nil
|
||||
end
|
||||
else
|
||||
raise SessionError, 'no login data was found in Login message'
|
||||
end
|
||||
if !user.nil? && user.authenticate(password)
|
||||
@log.debug "#{user} login via password"
|
||||
return user
|
||||
else
|
||||
@log.debug "#{username} login failure"
|
||||
return nil
|
||||
end
|
||||
else
|
||||
raise SessionError, 'no login data was found in Login message'
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
def access_music_session?(music_session_id, user)
|
||||
music_session = MusicSession.find_by_id(music_session_id)
|
||||
def access_music_session(music_session_id, user)
|
||||
music_session = MusicSession.find_by_id(music_session_id)
|
||||
|
||||
if music_session.nil?
|
||||
raise SessionError, 'specified session not found'
|
||||
end
|
||||
if music_session.nil?
|
||||
raise SessionError, 'specified session not found'
|
||||
end
|
||||
|
||||
if !music_session.access? user
|
||||
raise SessionError, 'not allowed to join the specified session'
|
||||
end
|
||||
if !music_session.access? user
|
||||
raise SessionError, 'not allowed to join the specified session'
|
||||
end
|
||||
|
||||
return music_session
|
||||
end
|
||||
return music_session
|
||||
end
|
||||
|
||||
def handle_session_directed(session_id, client_msg, client)
|
||||
# client_id = the id of the client being accessed
|
||||
# client = the current client
|
||||
def access_p2p(client_id, user, msg)
|
||||
|
||||
context = @clients[client]
|
||||
# ping_request and ping_ack messages are special in that they are simply allowed
|
||||
if msg.type == ClientMessage::Type::PING_REQUEST || msg.type == ClientMessage::Type::PING_ACK
|
||||
return nil
|
||||
end
|
||||
|
||||
# by not catching any exception here, this will kill the connection
|
||||
# if for some reason the client is trying to send to a session that it doesn't
|
||||
# belong to
|
||||
session = access_music_session?(session_id, context.user)
|
||||
music_session_client = MusicSessionClient.find_by_client_id(client_id)
|
||||
|
||||
@log.debug "publishing to session #{session} from client_id #{client.client_id}"
|
||||
# put it on the topic exchange for sessions
|
||||
@sessions_exchange.publish(client_msg.to_s, :routing_key => "session.#{session_id}", :properties => { :headers => { "client_id" => client.client_id } } )
|
||||
end
|
||||
if music_session_client.nil?
|
||||
raise PermissionError, 'specified client not found'
|
||||
end
|
||||
|
||||
def handle_user_directed(user, client_msg, client)
|
||||
if !music_session_client.access_p2p? user
|
||||
|
||||
raise SessionError, 'not implemented'
|
||||
end
|
||||
end
|
||||
raise SessionError, 'not allowed to message this client'
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
def handle_session_directed(session_id, client_msg, client)
|
||||
|
||||
context = @clients[client]
|
||||
|
||||
# by not catching any exception here, this will kill the connection
|
||||
# if for some reason the client is trying to send to a session that it doesn't
|
||||
# belong to
|
||||
session = access_music_session(session_id, context.user)
|
||||
|
||||
@log.debug "publishing to session #{session} from client_id #{client.client_id}"
|
||||
# put it on the topic exchange for sessions
|
||||
@sessions_exchange.publish(client_msg.to_s, :routing_key => "session.#{session_id}", :properties => {:headers => {"client_id" => client.client_id}})
|
||||
end
|
||||
|
||||
def handle_client_directed(to_client_id, client_msg, client)
|
||||
context = @clients[client]
|
||||
|
||||
# by not catching any exception here, a PermissionError will be thrown if this isn't valid
|
||||
# if for some reason the client is trying to send to a client that it doesn't
|
||||
# belong to
|
||||
access_p2p(to_client_id, context.user, client_msg)
|
||||
|
||||
# populate routing data
|
||||
client_msg.from = client.client_id
|
||||
|
||||
@log.debug "publishing to client #{to_client_id} from client_id #{client.client_id}"
|
||||
|
||||
# put it on the topic exchange for clients
|
||||
@clients_exchange.publish(client_msg.to_s, :routing_key => "client.#{to_client_id}", :properties => {:headers => {"client_id" => client.client_id}})
|
||||
end
|
||||
|
||||
def handle_user_directed(user, client_msg, client)
|
||||
|
||||
raise SessionError, 'not implemented'
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -2,78 +2,80 @@ require 'spec_helper'
|
|||
require 'thread'
|
||||
|
||||
LoginClient = Class.new do
|
||||
attr_accessor :onmsgblock, :onopenblock, :encode_json, :client_id
|
||||
attr_accessor :onmsgblock, :onopenblock, :encode_json, :client_id
|
||||
|
||||
def initiaize()
|
||||
def initiaize()
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
def onopen(&block)
|
||||
@onopenblock = block
|
||||
end
|
||||
def onopen(&block)
|
||||
@onopenblock = block
|
||||
end
|
||||
|
||||
def onmessage(&block)
|
||||
@onmsgblock = block
|
||||
end
|
||||
def onmessage(&block)
|
||||
@onmsgblock = block
|
||||
end
|
||||
|
||||
def close(&block)
|
||||
@oncloseblock = block
|
||||
end
|
||||
def close(&block)
|
||||
@oncloseblock = block
|
||||
end
|
||||
|
||||
def close_websocket()
|
||||
|
||||
end
|
||||
def close_websocket()
|
||||
|
||||
def send(msg)
|
||||
puts msg
|
||||
end
|
||||
end
|
||||
|
||||
def get_peername
|
||||
return "\x00\x02\x93\v\x7F\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00" # 37643, "localhost"
|
||||
end
|
||||
def send(msg)
|
||||
puts msg
|
||||
end
|
||||
|
||||
def get_peername
|
||||
return "\x00\x02\x93\v\x7F\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00" # 37643, "localhost"
|
||||
end
|
||||
|
||||
end
|
||||
|
||||
|
||||
# does a login and returns client
|
||||
def login(router, user, password)
|
||||
def login(router, user, password, client_id)
|
||||
|
||||
message_factory = MessageFactory.new
|
||||
client = LoginClient.new
|
||||
message_factory = MessageFactory.new
|
||||
client = LoginClient.new
|
||||
|
||||
login_ack = message_factory.login_ack("127.0.0.1")
|
||||
login_ack = message_factory.login_ack("127.0.0.1", client_id)
|
||||
|
||||
router.should_receive(:send_to_client).with(client, login_ack)
|
||||
client.should_receive(:onclose)
|
||||
client.should_receive(:onerror)
|
||||
client.should_receive(:request).and_return({ "query" => { "pb" => "true" } })
|
||||
client.should_receive(:get_peername).and_return("\x00\x02\x93\v\x7F\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00")
|
||||
router.should_receive(:send_to_client).with(client, login_ack)
|
||||
client.should_receive(:onclose)
|
||||
client.should_receive(:onerror)
|
||||
client.should_receive(:request).and_return({"query" => {"pb" => "true"}})
|
||||
client.should_receive(:get_peername).and_return("\x00\x02\x93\v\x7F\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00")
|
||||
|
||||
@router.new_client(client)
|
||||
client.onopenblock.call
|
||||
@router.new_client(client)
|
||||
client.onopenblock.call
|
||||
|
||||
# create a login message, and pass it into the router via onmsgblock.call
|
||||
login = message_factory.login_with_user_pass(user.email, password)
|
||||
# create a login message, and pass it into the router via onmsgblock.call
|
||||
login = message_factory.login_with_user_pass(user.email, password, :client_id => client_id)
|
||||
|
||||
# first log in
|
||||
client.onmsgblock.call login.to_s
|
||||
# first log in
|
||||
client.onmsgblock.call login.to_s
|
||||
|
||||
# then join music session
|
||||
return client
|
||||
# then join music session
|
||||
return client
|
||||
end
|
||||
|
||||
# currently commented out; we have deprecated logging in for jam sessions via websocket-gateway;
|
||||
# use rest API instead (or direct db access with factory-girl)
|
||||
def login_music_session(router, client, music_session)
|
||||
message_factory = MessageFactory.new
|
||||
login_music_session = message_factory.login_music_session(music_session.id)
|
||||
login_ack = message_factory.login_music_session_ack(false, nil);
|
||||
router.should_receive(:send_to_client).with(client, login_ack)
|
||||
client.onmsgblock.call login_music_session.to_s
|
||||
#message_factory = MessageFactory.new
|
||||
#login_music_session = message_factory.login_music_session(music_session.id)
|
||||
#login_ack = message_factory.login_music_session_ack(false, nil);
|
||||
#router.should_receive(:send_to_client).with(client, login_ack)
|
||||
#client.onmsgblock.call login_music_session.to_s
|
||||
end
|
||||
|
||||
|
||||
describe Router do
|
||||
|
||||
message_factory = MessageFactory.new
|
||||
message_factory = MessageFactory.new
|
||||
|
||||
before do
|
||||
|
||||
|
|
@ -100,31 +102,30 @@ describe Router do
|
|||
client.should_receive(:onerror)
|
||||
client.should_receive(:onmessage)
|
||||
client.should_receive(:encode_json=)
|
||||
client.should_receive(:client_id=)
|
||||
|
||||
@router.new_client(client)
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
describe "topic routing helpers" do
|
||||
it "create and delete user lookup set" do
|
||||
user = double(User)
|
||||
user.should_receive(:id).any_number_of_times.and_return("1")
|
||||
client = double("client")
|
||||
context = ClientContext.new(user, client)
|
||||
describe "topic routing helpers" do
|
||||
it "create and delete user lookup set" do
|
||||
user = double(User)
|
||||
user.should_receive(:id).any_number_of_times.and_return("1")
|
||||
client = double("client")
|
||||
context = ClientContext.new(user, client)
|
||||
|
||||
@router.user_context_lookup.length.should == 0
|
||||
@router.user_context_lookup.length.should == 0
|
||||
|
||||
@router.add_user(context)
|
||||
|
||||
@router.user_context_lookup.length.should == 1
|
||||
@router.add_user(context)
|
||||
|
||||
@router.remove_user(context)
|
||||
@router.user_context_lookup.length.should == 1
|
||||
|
||||
@router.user_context_lookup.length.should == 0
|
||||
end
|
||||
end
|
||||
@router.remove_user(context)
|
||||
|
||||
@router.user_context_lookup.length.should == 0
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
describe "login" do
|
||||
|
|
@ -145,8 +146,8 @@ describe Router do
|
|||
@onmsgblock = block
|
||||
end
|
||||
|
||||
def close_websocket()
|
||||
end
|
||||
def close_websocket()
|
||||
end
|
||||
|
||||
def close()
|
||||
end
|
||||
|
|
@ -160,7 +161,7 @@ describe Router do
|
|||
client.should_receive(:close_websocket)
|
||||
client.should_receive(:onclose)
|
||||
client.should_receive(:onerror)
|
||||
client.should_receive(:request).and_return({ "query" => { "pb" => "true" } })
|
||||
client.should_receive(:request).and_return({"query" => {"pb" => "true"}})
|
||||
|
||||
|
||||
@router.new_client(client)
|
||||
|
|
@ -178,7 +179,7 @@ describe Router do
|
|||
:password => "foobar", :password_confirmation => "foobar")
|
||||
@user.save
|
||||
|
||||
client1 = login(@router, @user, "foobar")
|
||||
client1 = login(@router, @user, "foobar", "1")
|
||||
end
|
||||
|
||||
|
||||
|
|
@ -190,40 +191,79 @@ describe Router do
|
|||
|
||||
music_session = FactoryGirl.create(:music_session, :creator => user1)
|
||||
|
||||
music_session_member1 = FactoryGirl.create(:music_session_client, :user => user1, :music_session => music_session)
|
||||
music_session_member2 = FactoryGirl.create(:music_session_client, :user => user2, :music_session => music_session)
|
||||
music_session_member1 = FactoryGirl.create(:music_session_client, :user => user1, :music_session => music_session, :client_id => "1")
|
||||
music_session_member2 = FactoryGirl.create(:music_session_client, :user => user2, :music_session => music_session, :client_id => "2")
|
||||
|
||||
# make a music_session and define two members
|
||||
|
||||
# create client 1, log him in, and log him in to music session
|
||||
client1 = login(@router, user1, "foobar")
|
||||
login_music_session(@router, client1, music_session)
|
||||
# create client 1, log him in, and log him in to music session
|
||||
client1 = login(@router, user1, "foobar", "1")
|
||||
login_music_session(@router, client1, music_session) \
|
||||
|
||||
end
|
||||
|
||||
it "should allow two valid subscribers to communicate with session-directed messages", :mq => true do
|
||||
it "should allow two valid subscribers to communicate with session-directed messages", :mq => true do
|
||||
|
||||
EventMachine.run do
|
||||
user1 = FactoryGirl.create(:user) # in the music session
|
||||
user2 = FactoryGirl.create(:user) # in the music session
|
||||
EventMachine.run do
|
||||
user1 = FactoryGirl.create(:user) # in the music session
|
||||
user2 = FactoryGirl.create(:user) # in the music session
|
||||
|
||||
music_session = FactoryGirl.create(:music_session, :creator => user1)
|
||||
|
||||
music_session_member1 = FactoryGirl.create(:music_session_client, :user => user1, :music_session => music_session)
|
||||
music_session_member2 = FactoryGirl.create(:music_session_client, :user => user2, :music_session => music_session)
|
||||
|
||||
# make a music_session and define two members
|
||||
music_session = FactoryGirl.create(:music_session, :creator => user1)
|
||||
|
||||
|
||||
# create client 1, log him in, and log him in to music session
|
||||
client1 = login(@router, user1, "foobar")
|
||||
login_music_session(@router, client1, music_session)
|
||||
# create client 1, log him in, and log him in to music session
|
||||
client1 = login(@router, user1, "foobar", "1")
|
||||
login_music_session(@router, client1, music_session)
|
||||
|
||||
client2 = login(@router, user2, "foobar")
|
||||
login_music_session(@router, client2, music_session)
|
||||
EM.stop
|
||||
end
|
||||
client2 = login(@router, user2, "foobar", "2")
|
||||
login_music_session(@router, client2, music_session)
|
||||
|
||||
# make a music_session and define two members
|
||||
|
||||
music_session_member1 = FactoryGirl.create(:music_session_client, :user => user1, :music_session => music_session, :client_id => "1")
|
||||
music_session_member2 = FactoryGirl.create(:music_session_client, :user => user2, :music_session => music_session, :client_id => "2")
|
||||
|
||||
|
||||
EM.stop
|
||||
end
|
||||
end
|
||||
|
||||
it "should allow two valid subscribers to communicate with p2p messages", :mq => true do
|
||||
|
||||
EventMachine.run do
|
||||
user1 = FactoryGirl.create(:user) # in the music session
|
||||
user2 = FactoryGirl.create(:user) # in the music session
|
||||
|
||||
music_session = FactoryGirl.create(:music_session, :creator => user1)
|
||||
|
||||
# create client 1, log him in, and log him in to music session
|
||||
client1 = login(@router, user1, "foobar", "1")
|
||||
#login_music_session(@router, client1, music_session)
|
||||
|
||||
client2 = login(@router, user2, "foobar", "2")
|
||||
#login_music_session(@router, client2, music_session)
|
||||
|
||||
# by creating
|
||||
music_session_member1 = FactoryGirl.create(:music_session_client, :user => user1, :music_session => music_session, :client_id => "1")
|
||||
|
||||
# now attempt to message p2p!
|
||||
|
||||
# first test: user 2 should be able to send a ping message to user 1, even though he isn't in the same session yet
|
||||
|
||||
# create a login message, and pass it into the router via onmsgblock.call
|
||||
ping = message_factory.ping_request("1", "2")
|
||||
|
||||
#@router.should_receive(:send_to_client) #.with(client1, ping)
|
||||
|
||||
## send ping to client 2
|
||||
#client2.onmsgblock.call ping.to_s
|
||||
|
||||
#music_session_member2 = FactoryGirl.create(:music_session_client, :user => user2, :music_session => music_session, :client_id => "2")
|
||||
|
||||
|
||||
EM.stop
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue