seth
/
jam-cloud
mirror of https://bitbucket.org/jamkazam/jam-cloud.git
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

* more websocket/heartbeat fixes

This commit is contained in:
Seth Call 2014-04-30 03:01:28 +00:00
parent e3e9cb5830
commit 9820562892
33 changed files with 454 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
db/manifest
View File

@ -146,3 +146,6 @@ bands_did_session.sql
email_change_default_sender.sql
affiliate_partners.sql
chat_messages.sql
diagnostics.sql
user_mods.sql
connection_stale_expire.sql

2
db/up/connection_stale_expire.sql Normal file
View File

@ -0,0 +1,2 @@
ALTER TABLE connections ADD COLUMN stale_time INTEGER NOT NULL DEFAULT 20;
ALTER TABLE connections ADD COLUMN expire_time INTEGER NOT NULL DEFAULT 30;

11
db/up/diagnostics.sql Normal file
View File

@ -0,0 +1,11 @@
CREATE TABLE diagnostics
(
id VARCHAR(64) NOT NULL DEFAULT uuid_generate_v4(),
user_id VARCHAR(64) NOT NULL REFERENCES users (id) ON DELETE CASCADE,
type VARCHAR(255) NOT NULL,
creator VARCHAR(255) NOT NULL,
data TEXT,
created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX diagnostics_type_idx ON diagnostics(type);

1
db/up/user_mods.sql Normal file
View File

@ -0,0 +1 @@
ALTER TABLE users ADD COLUMN mods JSON;

1
pb/src/client_container.proto
View File

@ -184,6 +184,7 @@ message LoginAck {
optional string music_session_id = 5; // the music session that the user was in very recently (likely due to dropped connection)
optional bool reconnected = 6; // if reconnect_music_session_id is specified, and the server could log the user into that session, then true is returned.
optional string user_id = 7; // the database user id
optional int32 connection_expire_time = 8; // this is how long the server gives you before killing your connection entirely after missing heartbeats
}
// route_to: server

2
ruby/lib/jam_ruby.rb
View File

@ -31,6 +31,7 @@ require "jam_ruby/lib/module_overrides"
require "jam_ruby/lib/s3_util"
require "jam_ruby/lib/s3_manager"
require "jam_ruby/lib/profanity"
require "jam_ruby/lib/json_validator"
require "jam_ruby/lib/em_helper.rb"
require "jam_ruby/lib/nav.rb"
require "jam_ruby/resque/audiomixer"
@ -75,6 +76,7 @@ require "jam_ruby/models/artifact_update"
require "jam_ruby/models/band_invitation"
require "jam_ruby/models/band_musician"
require "jam_ruby/models/connection"
require "jam_ruby/models/diagnostic"
require "jam_ruby/models/friendship"
require "jam_ruby/models/music_session"
require "jam_ruby/models/music_session_comment"

13
ruby/lib/jam_ruby/connection_manager.rb
View File

@ -129,7 +129,6 @@ WHERE
aasm_state = '#{Connection::CONNECT_STATE.to_s}'
RETURNING music_session_id
SQL
# @log.info("*** flag_connection_stale_with_client_id: client_id = #{client_id}; sql = #{sql}")
self.pg_conn.exec(sql) do |result|
# if we did update a client to stale, retriee music_session_id
@ -171,16 +170,16 @@ SQL
# NOTE this is only used for testing purposes;
# actual deletes will be processed in the websocket context which cleans up dependencies
def expire_stale_connections(max_seconds)
self.stale_connection_client_ids(max_seconds).each { |cid| self.delete_connection(cid) }
self.stale_connection_client_ids(max_seconds).each { |client| self.delete_connection(client[:client_id]) }
end
# expiring connections in stale state, which deletes them
def stale_connection_client_ids(max_seconds)
client_ids = []
clients = []
ConnectionManager.active_record_transaction do |connection_manager|
conn = connection_manager.pg_conn
sql =<<SQL
SELECT client_id, music_session_id, user_id FROM connections
SELECT client_id, music_session_id, user_id, client_type FROM connections
WHERE
updated_at < (NOW() - interval '#{max_seconds} second') AND
aasm_state = '#{Connection::STALE_STATE.to_s}'
@ -190,13 +189,15 @@ SQL
client_id = row['client_id']
music_session_id = row['music_session_id']
user_id = row['user_id']
client_type = row['client_type']
client_ids << client_id
clients << {client_id: client_id, music_session_id: music_session_id, client_type: client_type, user_id: user_id}
}
end
end
client_ids
clients
end

15
ruby/lib/jam_ruby/lib/json_validator.rb Normal file
View File

@ -0,0 +1,15 @@
# This needs to be outside the module to work.
class JsonValidator < ActiveModel::EachValidator
# implement the method called during validation
def is_json?(value)
begin
!!JSON.parse(value)
rescue
false
end
end
def validate_each(record, attribute, value)
record.errors[attribute] << 'must be JSON' unless value.nil? || is_json?(value)
end
end

5
ruby/lib/jam_ruby/message_factory.rb
View File

@ -53,7 +53,7 @@ module JamRuby
end
# create a login ack (login was successful)
def login_ack(public_ip, client_id, token, heartbeat_interval, music_session_id, reconnected, user_id)
def login_ack(public_ip, client_id, token, heartbeat_interval, music_session_id, reconnected, user_id, connection_expire_time)
login_ack = Jampb::LoginAck.new(
:public_ip => public_ip,
:client_id => client_id,
@ -61,7 +61,8 @@ module JamRuby
:heartbeat_interval => heartbeat_interval,
:music_session_id => music_session_id,
:reconnected => reconnected,
:user_id => user_id
:user_id => user_id,
:connection_expire_time => connection_expire_time
)
Jampb::ClientMessage.new(

85
ruby/lib/jam_ruby/models/diagnostic.rb Normal file
View File

@ -0,0 +1,85 @@
module JamRuby
class Diagnostic < ActiveRecord::Base
# occurs when the client does not see a heartbeat from the server in a while
NO_HEARTBEAT_ACK = 'NO_HEARTBEAT_ACK'
# occurs when the client sees the socket go down
WEBSOCKET_CLOSED_REMOTELY = 'WEBSOCKET_CLOSED_REMOTELY'
# occurs when the websocket-gateway has finally given up entirely on a connection with no heartbeats seen in a while
EXPIRED_STALE_CONNECTION = 'EXPIRED_STALE_CONNECTION'
# occurs when the websocket-gateway is trying to handle a heartbeat, but can't find any state for the user.
# this implies a coding error
MISSING_CLIENT_STATE = 'MISSING_CLIENT_STATE'
# websocket gateway did not recognize message. indicates out-of-date websocket-gateway
UNKNOWN_MESSAGE_TYPE = 'UNKNOWN_MESSAGE_TYPE'
# empty route_to in message; which is invalid. indicates programming error
MISSING_ROUTE_TO = 'MISSING_ROUTE_TO'
# websocket gateway got a client with the same client_id as an already-connected client
DUPLICATE_CLIENT = 'DUPLICATE_CLIENT'
DIAGNOSTIC_TYPES = [NO_HEARTBEAT_ACK, WEBSOCKET_CLOSED_REMOTELY, EXPIRED_STALE_CONNECTION,
MISSING_CLIENT_STATE, UNKNOWN_MESSAGE_TYPE, MISSING_ROUTE_TO,
DUPLICATE_CLIENT]
# creator types #
CLIENT = 'client'
WEBSOCKET_GATEWAY = 'websocket-gateway'
CREATORS = [CLIENT, WEBSOCKET_GATEWAY]
self.primary_key = 'id'
belongs_to :user, :inverse_of => :diagnostics, :class_name => "JamRuby::User", :foreign_key => "user_id"
validates :user, :presence => true
validates :type, :inclusion => {:in => DIAGNOSTIC_TYPES}
validates :creator, :inclusion => {:in => CREATORS}
validates :data, length: {maximum: 100000}
def self.expired_stale_connection(user, context_as_json)
Diagnostic.save(EXPIRED_STALE_CONNECTION, user, WEBSOCKET_GATEWAY, context_as_json) if user
end
def self.missing_client_state(user, context)
Diagnostic.save(MISSING_CLIENT_STATE, user, WEBSOCKET_GATEWAY, context.to_json) if user
end
def self.missing_connection(user, context)
Diagnostic.save(MISSING_CONNECTION, user, WEBSOCKET_GATEWAY, context.to_json) if user
end
def self.duplicate_client(user, context)
Diagnostic.save(DUPLICATE_CLIENT, user, WEBSOCKET_GATEWAY, context.to_json) if user
end
def self.unknown_message_type(user, client_msg)
Diagnostic.save(UNKNOWN_MESSAGE_TYPE, user, WEBSOCKET_GATEWAY, client_msg.to_json) if user
end
def self.missing_route_to(user, client_msg)
Diagnostic.save(MISSING_ROUTE_TO, user, WEBSOCKET_GATEWAY, client_msg.to_json) if user
end
def self.save(type, user, creator, data)
diagnostic = Diagnostic.new
if user.class == String
diagnostic.user_id = user
else
diagnostic.user = user
end
diagnostic.data = data
diagnostic.type = type
diagnostic.creator = creator
diagnostic.save
end
end
end

2
ruby/lib/jam_ruby/models/email_batch.rb
View File

@ -12,7 +12,7 @@ module JamRuby
VAR_FIRST_NAME = '@FIRSTNAME'
VAR_LAST_NAME = '@LASTNAME'
DEFAULT_SENDER = "support@jamkazam.com"
DEFAULT_SENDER = "noreply@jamkazam.com"
BATCH_SIZE = 1000
BODY_TEMPLATE =<<FOO

21
ruby/lib/jam_ruby/models/user.rb
View File

@ -17,7 +17,7 @@ module JamRuby
attr_accessible :first_name, :last_name, :email, :city, :password, :password_confirmation, :state, :country, :birth_date, :subscribe_email, :terms_of_service, :original_fpfile, :cropped_fpfile, :cropped_large_fpfile, :cropped_s3_path, :cropped_large_s3_path, :photo_url, :large_photo_url, :crop_selection, :lat, :lng
# updating_password corresponds to a lost_password
attr_accessor :updating_password, :updating_email, :updated_email, :update_email_confirmation_url, :administratively_created, :current_password, :setting_password, :confirm_current_password, :updating_avatar, :updating_progression_field
attr_accessor :updating_password, :updating_email, :updated_email, :update_email_confirmation_url, :administratively_created, :current_password, :setting_password, :confirm_current_password, :updating_avatar, :updating_progression_field, :mods_json
belongs_to :icecast_server_group, class_name: "JamRuby::IcecastServerGroup", inverse_of: :users, foreign_key: 'icecast_server_group_id'
@ -99,6 +99,9 @@ module JamRuby
# events
has_many :event_sessions, :class_name => "JamRuby::EventSession"
# diagnostics
has_many :diagnostics, :class_name => "JamRuby::Diagnostic"
# This causes the authenticate method to be generated (among other stuff)
#has_secure_password
@ -119,6 +122,7 @@ module JamRuby
validates :subscribe_email, :inclusion => {:in => [nil, true, false]}
validates :musician, :inclusion => {:in => [true, false]}
validates :show_whats_next, :inclusion => {:in => [nil, true, false]}
validates :mods, json: true
# custom validators
validate :validate_musician_instruments
@ -280,6 +284,19 @@ module JamRuby
self.music_sessions.size
end
# mods comes back as text; so give ourselves a parsed version
def mods_json
@mods_json ||= mods ? JSON.parse(mods, symbolize_names: true) : {}
end
def heartbeat_interval
mods_json[:heartbeat_interval]
end
def connection_expire_time
mods_json[:connection_expire_time]
end
def recent_history
recordings = Recording.where(:owner_id => self.id)
.order('created_at DESC')
@ -356,7 +373,7 @@ module JamRuby
return first_name + ' ' + last_name
end
return id
id
end
def set_password(old_password, new_password, new_password_confirmation)

6
ruby/spec/factories.rb
View File

@ -443,4 +443,10 @@ FactoryGirl.define do
message Faker::Lorem.characters(10)
end
end
factory :diagnostic, :class => JamRuby::Diagnostic do
type JamRuby::Diagnostic::NO_HEARTBEAT_ACK
creator JamRuby::Diagnostic::CLIENT
data Faker::Lorem.sentence
end
end

6
ruby/spec/jam_ruby/connection_manager_spec.rb
View File

@ -261,7 +261,11 @@ describe ConnectionManager do
cids = @connman.stale_connection_client_ids(1)
cids.size.should == 1
cids[0].should == client_id
cids[0][:client_id].should == client_id
cids[0][:client_type].should == 'native'
cids[0][:music_session_id].should be_nil
cids[0][:user_id].should == user_id
cids.each { |cid| @connman.delete_connection(cid) }
sleep(1)

18
ruby/spec/jam_ruby/models/diagnostic_spec.rb Normal file
View File

@ -0,0 +1,18 @@
require 'spec_helper'
describe Diagnostic do
let (:user) { FactoryGirl.create(:user) }
let (:diagnostic) { FactoryGirl.create(:diagnostic, user: user) }
it 'can be made' do
diagnostic.save!
end
it "validates type" do
diagnostic = FactoryGirl.build(:diagnostic, user: user, type: 'bleh')
diagnostic.errors[:type].should == []
end
end

42
ruby/spec/jam_ruby/models/user_spec.rb
View File

@ -21,6 +21,7 @@ describe User do
it { should respond_to(:admin) }
it { should respond_to(:valid_password?) }
it { should respond_to(:can_invite) }
it { should respond_to(:mods) }
it { should be_valid }
it { should_not be_admin }
@ -69,6 +70,24 @@ describe User do
it { should_not be_valid }
end
describe "when mods is null" do
before { @user.mods = nil }
it { should be_valid }
end
describe "when mods is empty" do
before { @user.mods = 'nil' }
it { should_not be_valid }
end
describe "when mods is json object" do
before { @user.mods = '{"key":"value"}' }
it { should be_valid }
end
describe "first or last name cant have profanity" do
it "should not let the first name have profanity" do
@user.first_name = "fuck you"
@ -429,6 +448,29 @@ describe User do
end
describe "mods" do
it "should allow update of JSON" do
@user.mods = {some_field: 5}.to_json
@user.save!
end
it "should return heartbeart interval" do
@user.heartbeat_interval_client.should be_nil
@user.mods = {heartbeat_interval_client: 5}.to_json
@user.save!
@user = User.find(@user.id) # necessary because mods_json is cached in the model
@user.heartbeat_interval_client.should == 5
end
it "should return connection_expire_time" do
@user.connection_expire_time.should be_nil
@user.mods = {connection_expire_time: 5}.to_json
@user.save!
@user = User.find(@user.id) # necessary because mods_json is cached in the model
@user.connection_expire_time.should == 5
end
end
=begin
describe "update avatar" do

52
web/app/assets/javascripts/AAA_Log.js
View File

@ -15,6 +15,12 @@
'exception', 'table'
];
var log_methods = {
'log':null, 'debug':null, 'info':null, 'warn':null, 'error':null, 'assert':null, 'trace':null, 'exception':null
}
var logCache = [];
if ('undefined' === typeof(context.console)) {
context.console = {};
$.each(console_methods, function(index, value) {
@ -27,23 +33,39 @@
context.console.debug = function() { console.log(arguments); }
}
context.JK.logger = context.console;
// http://tobyho.com/2012/07/27/taking-over-console-log/
function takeOverConsole(){
var console = window.console
if (!console) return
function intercept(method){
var original = console[method]
console[method] = function(){
// JW - some code to tone down logging. Uncomment the following, and
// then do your logging to logger.dbg - and it will be the only thing output.
// TODO - find a way to wrap this up so that debug logs can stay in, but this
// class can provide a way to enable/disable certain namespaces of logs.
/*
var fakeLogger = {};
$.each(console_methods, function(index, value) {
fakeLogger[value] = $.noop;
});
fakeLogger.dbg = function(m) {
context.console.debug(m);
};
context.JK.logger = fakeLogger;
*/
logCache.push([method].concat(arguments));
if(logCache.length > 50) {
// keep the cache size 50 or lower
logCache.pop();
}
if (original.apply){
// Do this for normal browsers
original.apply(console, arguments)
}else{
// Do this for IE
var message = Array.prototype.slice.apply(arguments).join(' ')
original(message)
}
}
}
var methods = ['log', 'warn', 'error']
for (var i = 0; i < methods.length; i++)
intercept(methods[i])
}
takeOverConsole();
context.JK.logger = context.console;
context.JK.logger.logCache = logCache;
})(window, jQuery);

30
web/app/assets/javascripts/JamServer.js
View File

@ -21,9 +21,11 @@
var lastHeartbeatSentTime = null;
var lastHeartbeatAckTime = null;
var lastHeartbeatFound = false;
var lastDisconnectedReason = null;
var heartbeatAckCheckInterval = null;
var notificationLastSeenAt = undefined;
var notificationLastSeen = undefined;
var clientClosedConnection = false;
// reconnection logic
var connectDeferred = null;
@ -53,6 +55,13 @@
server.socketClosedListeners = [];
server.connected = false;
var clientType = context.JK.clientType();
function heartbeatStateReset() {
lastHeartbeatSentTime = null;
lastHeartbeatAckTime = null;
lastHeartbeatFound = false;
}
// if activeElementVotes is null, then we are assuming this is the initial connect sequence
function initiateReconnect(activeElementVotes, in_error) {
@ -129,6 +138,7 @@
// this logic equates to 'if we have not received a heartbeat within heartbeatMissedMS, then get upset
if (new Date().getTime() - lastHeartbeatAckTime.getTime() > heartbeatMissedMS) {
logger.error("no heartbeat ack received from server after ", heartbeatMissedMS, " seconds . giving up on socket connection");
lastDisconnectedReason = 'NO_HEARTBEAT_ACK';
context.JK.JamServer.close(true);
}
else {
@ -163,6 +173,8 @@
connectTimeout = null;
}
heartbeatStateReset();
app.clientId = payload.client_id;
// tell the backend that we have logged in
@ -170,6 +182,7 @@
$.cookie('client_id', payload.client_id);
heartbeatMS = payload.heartbeat_interval * 1000;
logger.debug("jamkazam.js.loggedIn(): clientId now " + app.clientId + "; Setting up heartbeat every " + heartbeatMS + " MS");
heartbeatInterval = context.setInterval(_heartbeat, heartbeatMS);
@ -235,12 +248,19 @@
function performReconnect() {
rest.createDiagnostic({
type: lastDisconnectedReason,
data: {logs: logger.logCache, client_type: clientType, client_id: server.clientID}
});
if ($currentDisplay.is('.no-websocket-connection')) {
// this path is the 'not in session path'; so there is nothing else to do
$currentDisplay.hide();
// TODO: tell certain elements that we've reconnected
}
else {
// this path is the 'in session' path, where we actually reload the page
context.JK.CurrentSessionModel.leaveCurrentSession()
.always(function () {
window.location.reload();
@ -439,6 +459,7 @@
server.close = function (in_error) {
logger.log("closing websocket");
clientClosedConnection = true;
server.socket.close();
closedCleanup(in_error);
@ -447,7 +468,7 @@
server.rememberLogin = function () {
var token, loginMessage;
token = $.cookie("remember_token");
var clientType = context.jamClient.IsNativeClient() ? 'client' : 'browser';
loginMessage = msg_factory.login_with_token(token, null, clientType);
server.send(loginMessage);
};
@ -483,13 +504,18 @@
}
};
// onClose is called if either client or server closes connection
server.onClose = function () {
logger.log("Socket to server closed.");
logger.log("Socket to server closed.", arguments);
if (connectDeferred.state() === "pending") {
connectDeferred.reject();
}
if(!clientClosedConnection) {
lastDisconnectedReason = 'WEBSOCKET_CLOSED_REMOTELY'
clientClosedConnection = false;
}
closedCleanup(true);
};

11
web/app/assets/javascripts/jam_rest.js
View File

@ -944,6 +944,16 @@
});
}
function createDiagnostic(options) {
return $.ajax({
type: "POST",
url: '/api/diagnostics',
dataType: "json",
contentType: 'application/json',
data: JSON.stringify(options)
});
}
function initialize() {
return self;
}
@ -1026,6 +1036,7 @@
this.createFbInviteUrl = createFbInviteUrl;
this.createTextMessage = createTextMessage;
this.getNotifications = getNotifications;
this.createDiagnostic = createDiagnostic;
return this;
};

3
web/app/assets/javascripts/utils.js
View File

@ -572,6 +572,9 @@
doneYet();
};
context.JK.clientType = function () {
return context.jamClient.IsNativeClient() ? 'client' : 'browser';
}
/**
* Returns 'MacOSX' if the os appears to be macintosh,
* 'Win32' if the os appears to be windows,

16
web/app/controllers/api_diagnostics_controller.rb Normal file
View File

@ -0,0 +1,16 @@
class ApiDiagnosticsController < ApiController
before_filter :api_signed_in_user
respond_to :json
def create
@diagnostic = Diagnostic.new
@diagnostic.type = params[:type]
@diagnostic.data = params[:data].to_json if params[:data]
@diagnostic.user = current_user
@diagnostic.creator = Diagnostic::CLIENT
@diagnostic.save
respond_with_model(@diagnostic, new: true)
end
end

2
web/config/initializers/eventmachine.rb
View File

@ -10,7 +10,7 @@ unless $rails_rake_task
:port => APP_CONFIG.websocket_gateway_port,
:emwebsocket_debug => APP_CONFIG.websocket_gateway_internal_debug,
:connect_time_stale => APP_CONFIG.websocket_gateway_connect_time_stale,
:connect_time_expire => APP_CONFIG.websocket_gateway_connect_time_expire,
:connect_time_expire_client => APP_CONFIG.websocket_gateway_connect_time_expire,
:rabbitmq_host => APP_CONFIG.rabbitmq_host,
:rabbitmq_port => APP_CONFIG.rabbitmq_port,
:calling_thread => current)

3
web/config/routes.rb
View File

@ -402,6 +402,9 @@ SampleApp::Application.routes.draw do
# favorites
match '/favorites' => 'api_favorites#index', :via => :get
match '/favorites/:id' => 'api_favorites#update', :via => :post
# diagnostic
match '/diagnostics' => 'api_diagnostics#create', :via => :post
end
end

43
web/spec/requests/diagnostics_api_spec.rb Normal file
View File

@ -0,0 +1,43 @@
require 'spec_helper'
# user progression is achieved by different aspects of the code working together in a cross-cutting fashion.
# due to this, it's nice to have a single place where all the parts of user progression are tested
# https://jamkazam.atlassian.net/wiki/pages/viewpage.action?pageId=3375145
describe "Diagnostics", :type => :api do
include Rack::Test::Methods
let(:user) { FactoryGirl.create(:user) }
subject { page }
def login(user)
post '/sessions', "session[email]" => user.email, "session[password]" => user.password
rack_mock_session.cookie_jar["remember_token"].should == user.remember_token
end
describe "create" do
before do
Diagnostic.delete_all
login(user)
end
it "can fail" do
post "/api/diagnostics.json", {}.to_json, "CONTENT_TYPE" => 'application/json'
last_response.status.should eql(422)
JSON.parse(last_response.body).should eql({"errors"=>{"type"=>["is not included in the list"], "creator"=>["is not included in the list"]}})
Diagnostic.count.should == 0
end
it "can succeed" do
post "/api/diagnostics.json", { type: Diagnostic::NO_HEARTBEAT_ACK}.to_json, "CONTENT_TYPE" => 'application/json'
last_response.status.should eql(201)
Diagnostic.count.should == 1
end
end
end

2
web/spec/spec_helper.rb
View File

@ -76,7 +76,7 @@ Thread.new do
:port => 6769,
:emwebsocket_debug => false,
:connect_time_stale => 2,
:connect_time_expire => 5,
:connect_time_expire_client => 5,
:rabbitmq_host => 'localhost',
:rabbitmq_port => 5672,
:calling_thread => current)

2
web/vendor/assets/javascripts/jquery.icheck.js vendored
View File

@ -285,12 +285,10 @@
// Check, disable or indeterminate
if (/^(ch|di|in)/.test(method) && !active) {
console.log("TAKING ROUTE: ", state);
on(input, state);
// Uncheck, enable or determinate
} else if (/^(un|en|de)/.test(method) && active) {
console.log("TAKING ROUTE2: ", state);
off(input, state);
// Update

6
websocket-gateway/bin/websocket_gateway
View File

@ -47,7 +47,9 @@ Object.send(:remove_const, :Rails) # this is to 'fool' new relic into not thinki
Server.new.run(:port => config["port"],
:emwebsocket_debug => config["emwebsocket_debug"],
:connect_time_stale => config["connect_time_stale"],
:connect_time_expire => config["connect_time_expire"],
:connect_time_stale_client => config["connect_time_stale_client"],
:connect_time_expire_client => config["connect_time_expire_client"],
:connect_time_stale_browser => config["connect_time_stale_browser"],
:connect_time_expire_browser => config["connect_time_expire_browser"],
:rabbitmq_host => config['rabbitmq_host'],
:rabbitmq_port => config['rabbitmq_port'])

6
websocket-gateway/config/application.yml
View File

@ -1,6 +1,8 @@
Defaults: &defaults
connect_time_stale: 6
connect_time_expire: 10
connect_time_stale_client: 20
connect_time_expire_client: 30
connect_time_stale_browser: 40
connect_time_expire_browser: 60
development:
port: 6767

11
websocket-gateway/lib/jam_websockets/client_context.rb
View File

@ -1,20 +1,27 @@
module JamWebsockets
class ClientContext
attr_accessor :user, :client, :msg_count, :session, :sent_bad_state_previously
attr_accessor :user, :client, :msg_count, :session, :client_type, :sent_bad_state_previously
def initialize(user, client)
def initialize(user, client, client_type)
@user = user
@client = client
@client_type = client_type
@msg_count = 0
@session = nil
@sent_bad_state_previously = false
client.context = self
end
def to_s
return "Client[user:#{@user} client:#{@client} msgs:#{@msg_count} session:#{@session}]"
end
def to_json
{user_id: @user.id, client_id: @client.client_id, msg_count: @msg_count, client_type: @client_type}.to_json
end
def hash
@client.hash
end

107
websocket-gateway/lib/jam_websockets/router.rb
View File

@ -10,7 +10,7 @@ include Jampb
module EventMachine
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
attr_accessor :encode_json, :client_id, :user_id, :context # client_id is uuid we give to each client to track them as we like
# http://stackoverflow.com/questions/11150147/how-to-check-if-eventmachineconnection-is-open
attr_accessor :connected
@ -39,7 +39,6 @@ module JamWebsockets
def initialize()
@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
@client_lookup = {} # lookup a client by client_id
@ -50,15 +49,19 @@ module JamWebsockets
@user_topic = nil
@client_topic = nil
@thread_pool = nil
@heartbeat_interval = nil
@heartbeat_interval_client = nil
@connect_time_expire_client = nil
@heartbeat_interval_browser= nil
@connect_time_expire_browser= nil
@ar_base_logger = ::Logging::Repository.instance[ActiveRecord::Base]
end
def start(connect_time_stale, options={:host => "localhost", :port => 5672}, &block)
def start(connect_time_stale_client, connect_time_expire_client, connect_time_stale_browser, connect_time_expire_browser, options={:host => "localhost", :port => 5672}, &block)
@log.info "startup"
@heartbeat_interval = connect_time_stale / 2
@heartbeat_interval_client = connect_time_stale_client / 2
@connect_time_expire_client = connect_time_expire_client
begin
@amqp_connection_manager = AmqpConnectionManager.new(true, 4, :host => options[:host], :port => options[:port])
@ -214,10 +217,6 @@ module JamWebsockets
def new_client(client)
@semaphore.synchronize do
@pending_clients.add(client)
end
# default to using json instead of pb
client.encode_json = true
@ -237,6 +236,7 @@ module JamWebsockets
client.onclose {
@log.debug "Connection closed"
stale_client(client)
cleanup_client(client)
}
client.onerror { |error|
@ -271,7 +271,6 @@ module JamWebsockets
error_msg = @message_factory.server_rejection_error(e.to_s)
send_to_client(client, error_msg)
ensure
client.close
cleanup_client(client)
end
rescue PermissionError => e
@ -290,7 +289,6 @@ module JamWebsockets
error_msg = @message_factory.server_generic_error(e.to_s)
send_to_client(client, error_msg)
ensure
client.close
cleanup_client(client)
end
end
@ -328,9 +326,9 @@ module JamWebsockets
# caused a client connection to be marked stale
def stale_client(client)
if cid = client.client_id
if client.client_id
ConnectionManager.active_record_transaction do |connection_manager|
music_session_id = connection_manager.flag_connection_stale_with_client_id(cid)
music_session_id = connection_manager.flag_connection_stale_with_client_id(client.client_id)
# update the session members, letting them know this client went stale
context = @client_lookup[client.client_id]
if music_session = MusicSession.find_by_id(music_session_id)
@ -340,12 +338,14 @@ module JamWebsockets
end
end
def cleanup_clients_with_ids(client_ids)
# @log.debug("*** cleanup_clients_with_ids: client_ids = #{client_ids.inspect}")
client_ids.each do |cid|
def cleanup_clients_with_ids(expired_connections)
expired_connections.each do |expired_connection|
cid = expired_connection[:client_id]
client_context = @client_lookup[cid]
self.cleanup_client(client_context.client) unless client_context.nil?
diagnostic_data = client_context.to_json unless client_context.nil?
cleanup_client(client_context.client) unless client_context.nil?
music_session = nil
recordingId = nil
@ -354,6 +354,7 @@ module JamWebsockets
# remove this connection from the database
ConnectionManager.active_record_transaction do |mgr|
mgr.delete_connection(cid) { |conn, count, music_session_id, user_id|
Diagnostic.expired_stale_connection(user_id, diagnostic_data)
Notification.send_friend_update(user_id, false, conn) if count == 0
music_session = MusicSession.find_by_id(music_session_id) unless music_session_id.nil?
user = User.find_by_id(user_id) unless user_id.nil?
@ -374,8 +375,7 @@ module JamWebsockets
@semaphore.synchronize do
client.close if client.connected?
# @log.debug("*** cleanup_clients: client = #{client}")
pending = @pending_clients.delete?(client)
pending = client.context.nil? # presence of context implies this connection has been logged into
if pending
@log.debug "cleaned up not-logged-in client #{client}"
@ -387,7 +387,7 @@ module JamWebsockets
remove_client(client.client_id)
remove_user(context)
else
@log.debug "skipping duplicate cleanup attempt of logged-in client"
@log.warn "skipping duplicate cleanup attempt of logged-in client"
end
end
end
@ -395,14 +395,19 @@ module JamWebsockets
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?
if message_type.nil?
Diagnostic.unknown_message_type(client.user_id, client_msg)
raise SessionError, "unknown message type received: #{client_msg.type}" if message_type.nil?
end
@log.debug("msg received #{message_type}") if client_msg.type != ClientMessage::Type::HEARTBEAT
raise SessionError, 'client_msg.route_to is null' if client_msg.route_to.nil?
if client_msg.route_to.nil?
Diagnostic.missing_route_to(client.user_id, client_msg)
raise SessionError, 'client_msg.route_to is null'
end
if @pending_clients.include? client and client_msg.type != ClientMessage::Type::LOGIN
if !client.user_id and client_msg.type != ClientMessage::Type::LOGIN
# this client has not logged in and is trying to send a non-login message
raise SessionError, "must 'Login' first"
end
@ -463,25 +468,28 @@ module JamWebsockets
user = valid_login(username, password, token, client_id)
# kill any websocket connections that have this same client_id, which can happen in race conditions
existing_client = @client_lookup[client_id]
if existing_client
remove_client(client_id)
existing_client.client.close
# this code must happen here, before we go any further, so that there is only one websocket connection per client_id
existing_context = @client_lookup[client_id]
if existing_context
# in reconnect scenarios, we may have in memory a client still
Diagnostic.duplicate_client(existing_context.user, existing_context) if existing_context.client.connected
cleanup_client(existing_context.client)
end
connection = JamRuby::Connection.find_by_client_id(client_id)
# if this connection is reused by a different user, then whack the connection
# because it will recreate a new connection lower down
if !connection.nil? && !user.nil? && connection.user != user
if connection && user && connection.user != user
@log.debug("user #{user.email} took client_id #{client_id} from user #{connection.user.email}")
connection.delete
connection = nil
end
client.client_id = client_id
client.user_id = user.id if user
remote_ip = extract_ip(client)
if !user.nil?
if user
@log.debug "user #{user} logged in with client_id #{client_id}"
# check if there's a connection for the client... if it's stale, reconnect it
@ -492,19 +500,17 @@ module JamWebsockets
music_session_upon_reentry = connection.music_session
send_depart = false
recordingId = nil
context = nil
recording_id = nil
ConnectionManager.active_record_transaction do |connection_manager|
music_session_id, reconnected = connection_manager.reconnect(connection, reconnect_music_session_id, remote_ip)
context = @client_lookup[client_id]
if music_session_id.nil?
# if this is a reclaim of a connection, but music_session_id comes back null, then we need to check if this connection was IN a music session before.
# if so, then we need to tell the others in the session that this user is now departed
unless context.nil? || music_session_upon_reentry.nil? || music_session_upon_reentry.destroyed?
unless music_session_upon_reentry.nil? || music_session_upon_reentry.destroyed?
recording = music_session_upon_reentry.stop_recording
recordingId = recording.id unless recording.nil?
recording_id = recording.id unless recording.nil?
music_session_upon_reentry.with_lock do # VRFS-1297
music_session_upon_reentry.tick_track_changes
end
@ -512,24 +518,21 @@ module JamWebsockets
end
else
music_session = MusicSession.find_by_id(music_session_id)
Notification.send_musician_session_fresh(music_session, client.client_id, context.user) unless context.nil?
Notification.send_musician_session_fresh(music_session, client.client_id, user)
end
end if connection.stale?
end
if send_depart
Notification.send_session_depart(music_session_upon_reentry, client.client_id, context.user, recordingId)
Notification.send_session_depart(music_session_upon_reentry, client.client_id, user, recording_id)
end
end
# respond with LOGIN_ACK to let client know it was successful
@semaphore.synchronize do
# remove from pending_queue
@pending_clients.delete(client)
# add a tracker for this user
context = ClientContext.new(user, client)
context = ClientContext.new(user, client, client_type)
@clients[client] = context
add_user(context)
add_client(client_id, context)
@ -544,13 +547,21 @@ module JamWebsockets
end
end
end
heartbeat_interval = user.heartbeat_interval_client.to_i || @heartbeat_interval_client
heartbeat_interval = @heartbeat_interval_client if heartbeat_interval == 0 # protect against bad config
connection_expire_time = user.connection_expire_time || @connection_expire_time
connection_expire_time = @connection_expire_time if connection_expire_time == 0 # protect against bad config
login_ack = @message_factory.login_ack(remote_ip,
client_id,
user.remember_token,
@heartbeat_interval,
@heartbeat_interval_client,
connection.try(:music_session_id),
reconnected,
user.id)
user.id,
@connection_expire_time)
send_to_client(client, login_ack)
end
else
@ -560,15 +571,15 @@ module JamWebsockets
def handle_heartbeat(heartbeat, heartbeat_message_id, client)
unless context = @clients[client]
@log.warn "*** WARNING: unable to find context due to heartbeat from client: #{client.client_id}; calling cleanup"
cleanup_client(client)
@log.warn "*** WARNING: unable to find context when handling heartbeat. client_id=#{client.client_id}; killing session"
Diagnostic.missing_client_state(client.user_id, client.context)
raise SessionError, 'context state is gone. please reconnect.'
else
connection = Connection.find_by_user_id_and_client_id(context.user.id, context.client.client_id)
track_changes_counter = nil
if connection.nil?
@log.warn "*** WARNING: unable to find connection due to heartbeat from client: #{context}; calling cleanup_client"
cleanup_client(client)
@log.warn "*** WARNING: unable to find connection when handling heartbeat. context= #{context}; killing session"
Diagnostic.missing_connection(client.user_id, client.context)
raise SessionError, 'connection state is gone. please reconnect.'
else
Connection.transaction do

24
websocket-gateway/lib/jam_websockets/server.rb
View File

@ -16,13 +16,15 @@ module JamWebsockets
def run(options={})
host = "0.0.0.0"
port = options[:port]
connect_time_stale = options[:connect_time_stale].to_i
connect_time_expire = options[:connect_time_expire].to_i
connect_time_stale_client = options[:connect_time_stale_client].to_i
connect_time_expire_client = options[:connect_time_expire_client].to_i
connect_time_stale_browser = options[:connect_time_stale_browser].to_i
connect_time_expire_browser = options[:connect_time_expire_browser].to_i
rabbitmq_host = options[:rabbitmq_host]
rabbitmq_port = options[:rabbitmq_port].to_i
calling_thread = options[:calling_thread]
@log.info "starting server #{host}:#{port} staleness_time=#{connect_time_stale}; reconnect time = #{connect_time_expire}, rabbitmq=#{rabbitmq_host}:#{rabbitmq_port}"
@log.info "starting server #{host}:#{port} staleness_time=#{connect_time_stale_client}; reconnect time = #{connect_time_expire_client}, rabbitmq=#{rabbitmq_host}:#{rabbitmq_port}"
EventMachine.error_handler{|e|
@log.error "unhandled error #{e}"
@ -30,13 +32,10 @@ module JamWebsockets
}
EventMachine.run do
@router.start(connect_time_stale, host: rabbitmq_host, port: rabbitmq_port) do
# take stale off the expire limit because the call to stale will
# touch the updated_at column, adding an extra stale limit to the expire time limit
# expire_time = connect_time_expire > connect_time_stale ? connect_time_expire - connect_time_stale : connect_time_expire
expire_time = connect_time_expire
@router.start(connect_time_stale_client, connect_time_expire_client, connect_time_stale_browser, connect_time_expire_browser, host: rabbitmq_host, port: rabbitmq_port) do
expire_time = connect_time_expire_client
start_connection_expiration(expire_time)
start_connection_flagger(connect_time_stale)
start_connection_flagger(connect_time_stale_client)
start_websocket_listener(host, port, options[:emwebsocket_debug])
calling_thread.wakeup if calling_thread
end
@ -72,12 +71,11 @@ module JamWebsockets
end
def expire_stale_connections(stale_max_time)
client_ids = []
clients = []
ConnectionManager.active_record_transaction do |connection_manager|
client_ids = connection_manager.stale_connection_client_ids(stale_max_time)
clients = connection_manager.stale_connection_client_ids(stale_max_time)
end
# @log.debug("*** expire_stale_connections(#{stale_max_time}): client_ids = #{client_ids.inspect}")
@router.cleanup_clients_with_ids(client_ids)
@router.cleanup_clients_with_ids(clients)
end
def start_connection_flagger(flag_max_time)

2
websocket-gateway/spec/jam_websockets/client_context_spec.rb
View File

@ -2,7 +2,7 @@ require 'spec_helper'
describe ClientContext do
let(:context) {ClientContext.new({}, "client1")}
let(:context) {ClientContext.new({}, "client1", "client")}
describe 'hashing' do
it "hash correctly" do

4
websocket-gateway/spec/jam_websockets/router_spec.rb
View File

@ -42,7 +42,7 @@ def login(router, user, password, client_id)
message_factory = MessageFactory.new
client = LoginClient.new
login_ack = message_factory.login_ack("127.0.0.1", client_id, user.remember_token, 15, nil, false, user.id)
login_ack = message_factory.login_ack("127.0.0.1", client_id, user.remember_token, 15, nil, false, user.id, 30)
router.should_receive(:send_to_client) do |*args|
args.count.should == 2
@ -126,7 +126,7 @@ describe Router do
user = double(User)
user.should_receive(:id).any_number_of_times.and_return("1")
client = double("client")
context = ClientContext.new(user, client)
context = ClientContext.new(user, client, "client")
@router.user_context_lookup.length.should == 0