From 422668a67f8607b7da851fef33ff1b267bdad632 Mon Sep 17 00:00:00 2001 From: Pratik Vyas Date: Sat, 20 Jun 2015 21:10:10 +0530 Subject: [PATCH 01/47] Async --- frappe/__init__.py | 2 + frappe/api.py | 4 +- frappe/app.py | 27 +- frappe/async.py | 175 + frappe/build.py | 3 +- frappe/celery_app.py | 3 +- frappe/core/doctype/async_task/__init__.py | 0 .../core/doctype/async_task/async_task.json | 142 + frappe/core/doctype/async_task/async_task.py | 10 + .../doctype/async_task/test_async_task.py | 12 + frappe/handler.py | 16 +- frappe/hooks.py | 1 + frappe/installer.py | 3 +- frappe/public/build.json | 5 +- frappe/public/js/frappe/request.js | 13 +- frappe/public/js/frappe/socket.js | 60 + frappe/public/js/lib/socket.io.min.js | 7000 +++++++++++++++++ frappe/sessions.py | 2 +- frappe/tasks.py | 47 + frappe/tests/test_async.py | 16 + frappe/utils/__init__.py | 1 + requirements.txt | 1 + socketio.js | 67 + 23 files changed, 7600 insertions(+), 10 deletions(-) create mode 100644 frappe/async.py create mode 100644 frappe/core/doctype/async_task/__init__.py create mode 100644 frappe/core/doctype/async_task/async_task.json create mode 100644 frappe/core/doctype/async_task/async_task.py create mode 100644 frappe/core/doctype/async_task/test_async_task.py create mode 100644 frappe/public/js/frappe/socket.js create mode 100644 frappe/public/js/lib/socket.io.min.js create mode 100644 frappe/tests/test_async.py create mode 100644 socketio.js diff --git a/frappe/__init__.py b/frappe/__init__.py index f5b5c29b64..877e109281 100644 --- a/frappe/__init__.py +++ b/frappe/__init__.py @@ -7,6 +7,7 @@ globals attached to frappe module from __future__ import unicode_literals from werkzeug.local import Local, release_local +from functools import wraps import os, importlib, inspect, logging, json # public @@ -14,6 +15,7 @@ from frappe.__version__ import __version__ from .exceptions import * from .utils.jinja import get_jenv, get_template, render_template + local = Local() class _dict(dict): diff --git a/frappe/api.py b/frappe/api.py index 425d48725c..5ec69d5f6d 100644 --- a/frappe/api.py +++ b/frappe/api.py @@ -58,13 +58,13 @@ def handle(): if frappe.local.request.method=="GET": if not doc.has_permission("read"): frappe.throw(_("Not permitted"), frappe.PermissionError) - doc.run_method(method, **frappe.local.form_dict) + frappe.local.response.update({"data": doc.run_method(method, **frappe.local.form_dict)}) if frappe.local.request.method=="POST": if not doc.has_permission("write"): frappe.throw(_("Not permitted"), frappe.PermissionError) - doc.run_method(method, **frappe.local.form_dict) + frappe.local.response.update({"data": doc.run_method(method, **frappe.local.form_dict)}) frappe.db.commit() else: diff --git a/frappe/app.py b/frappe/app.py index d670cb8e56..adaabc8da6 100644 --- a/frappe/app.py +++ b/frappe/app.py @@ -12,6 +12,8 @@ from werkzeug.local import LocalManager from werkzeug.exceptions import HTTPException, NotFound from werkzeug.contrib.profiler import ProfilerMiddleware from werkzeug.wsgi import SharedDataMiddleware +from werkzeug.serving import run_with_reloader + import mimetypes import frappe @@ -20,9 +22,10 @@ import frappe.auth import frappe.api import frappe.utils.response import frappe.website.render -from frappe.utils import get_site_name +from frappe.utils import get_site_name, get_site_path from frappe.middlewares import StaticDataMiddleware + local_manager = LocalManager([frappe.local]) _site = None @@ -30,6 +33,21 @@ _sites_path = os.environ.get("SITES_PATH", ".") logger = frappe.get_logger() +class RequestContext(object): + + def __init__(self, environ): + self.request = Request(environ) + + def __enter__(self): + frappe.local.request = self.request + init_site(self.request) + make_form_dict(self.request) + frappe.local.http_request = frappe.auth.HTTPRequest() + + def __exit__(self, type, value, traceback): + frappe.destroy() + + @Request.application def application(request): frappe.local.request = request @@ -135,6 +153,8 @@ def make_form_dict(request): frappe.local.form_dict.pop("_") application = local_manager.make_middleware(application) +application.debug = True + def serve(port=8000, profile=False, site=None, sites_path='.'): global application, _site, _sites_path @@ -155,5 +175,10 @@ def serve(port=8000, profile=False, site=None, sites_path='.'): b'/files': os.path.abspath(sites_path).encode("utf-8") }) + application.debug = True + application.config = { + 'SERVER_NAME': 'localhost:8000' + } + run_simple('0.0.0.0', int(port), application, use_reloader=True, use_debugger=True, use_evalex=True, threaded=True) diff --git a/frappe/async.py b/frappe/async.py new file mode 100644 index 0000000000..48ebdcea76 --- /dev/null +++ b/frappe/async.py @@ -0,0 +1,175 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and contributors +# For license information, please see license.txt + +from __future__ import unicode_literals + + +import frappe +import os +import time +from functools import wraps +from frappe.utils import get_site_path +import json +from frappe import conf + +END_LINE = '' +TASK_LOG_MAX_AGE = 86400 # 1 day in seconds +redis_server = None + + +def handler(f): + cmd = f.__module__ + '.' + f.__name__ + + def _run(args, set_in_response=True): + from frappe.tasks import run_async_task + args = frappe._dict(args) + task = run_async_task.delay(frappe.local.site, + (frappe.session and frappe.session.user) or 'Administrator', cmd, args) + if set_in_response: + frappe.local.response['task_id'] = task.id + return task.id + + @wraps(f) + def _f(*args, **kwargs): + from frappe.tasks import run_async_task + task = run_async_task.delay(frappe.local.site, + (frappe.session and frappe.session.user) or 'Administrator', cmd, + frappe.local.form_dict) + frappe.local.response['task_id'] = task.id + return { + "status": "queued", + "task_id": task.id + } + _f.async = True + _f._f = f + _f.run = _run + frappe.whitelisted.append(f) + frappe.whitelisted.append(_f) + return _f + + +def run_async_task(method, args, reference_doctype=None, reference_name=None, set_in_response=True): + if frappe.local.request and frappe.local.request.method == "GET": + frappe.throw("Cannot run task in a GET request") + task_id = method.run(args, set_in_response=set_in_response) + task = frappe.new_doc("Async Task") + task.celery_task_id = task_id + task.status = "Queued" + task.reference_doctype = reference_doctype + task.reference_name = reference_name + task.save() + return task_id + + +@frappe.whitelist() +def get_pending_tasks_for_doc(doctype, docname): + return frappe.db.sql_list("select name from `tabAsync Task` where status in ('Queued', 'Running') and reference_doctype='%s' and reference_name='%s'" % (doctype, docname)) + + +@handler +def ping(): + from time import sleep + sleep(6) + return "pong" + + +@frappe.whitelist() +def get_task_status(task_id): + from frappe.celery_app import get_celery + c = get_celery() + a = c.AsyncResult(task_id) + frappe.local.response['response'] = a.result + return { + "state": a.state, + "progress": 0 + } + + +def set_task_status(task_id, status, response=None): + frappe.db.set_value("Async Task", task_id, "status", status) + if not response: + response = {} + response.update({ + "status": status, + "task_id": task_id + }) + emit_via_redis("task_status_change", response, room="task:" + task_id) + + +def remove_old_task_logs(): + logs_path = get_site_path('task-logs') + + def full_path(_file): + return os.path.join(logs_path, _file) + + files_to_remove = [full_path(_file) for _file in os.listdir(logs_path)] + files_to_remove = [_file for _file in files_to_remove if is_file_old(_file) and os.path.isfile(_file)] + for _file in files_to_remove: + os.remove(_file) + + +def is_file_old(file_path): + return ((time.time() - os.stat(file_path).st_mtime) > TASK_LOG_MAX_AGE) + + +def emit_via_redis(event, message, room=None): + r = get_redis_server() + r.publish('events', json.dumps({'event': event, 'message': message, 'room': room})) + + +def put_log(task_id, line_no, line): + r = get_redis_server() + print "task_log:" + task_id + r.hset("task_log:" + task_id, line_no, line) + + +def get_redis_server(): + """Returns memcache connection.""" + global redis_server + if not redis_server: + from redis import Redis + redis_server = Redis.from_url(conf.get("cache_redis_server") or "redis://localhost:12311") + return redis_server + + +class FileAndRedisStream(file): + def __init__(self, *args, **kwargs): + ret = super(FileAndRedisStream, self).__init__(*args, **kwargs) + self.count = 0 + return ret + + def write(self, data): + ret = super(FileAndRedisStream, self).write(data) + if frappe.local.task_id: + emit_via_redis('task_progress', { + "message": { + "lines": {self.count: data} + }, + "task_id": frappe.local.task_id + }, room="task_progress:" + frappe.local.task_id) + + put_log(frappe.local.task_id, self.count, data) + self.count += 1 + return ret + + +def get_std_streams(task_id): + stdout = FileAndRedisStream(get_task_log_file_path(task_id, 'stdout'), 'w') + # stderr = FileAndRedisStream(get_task_log_file_path(task_id, 'stderr'), 'w') + return stdout, stdout + + +def get_task_log_file_path(task_id, stream_type): + logs_dir = frappe.utils.get_site_path('task-logs') + return os.path.join(logs_dir, task_id + '.' + stream_type) + + +@frappe.whitelist(allow_guest=True) +def can_subscribe_doc(doctype, docname, sid): + from frappe.sessions import Session + from frappe.exceptions import PermissionError + session = Session(None).get_session_data() + if not frappe.has_permission(user=session.user, doctype=doctype, doc=docname, ptype='read'): + raise PermissionError() + return True diff --git a/frappe/build.py b/frappe/build.py index 42136f59fa..d65efc538a 100644 --- a/frappe/build.py +++ b/frappe/build.py @@ -127,7 +127,8 @@ def pack(target, sources, no_compress, verbose): tmpin, tmpout = StringIO(data.encode('utf-8')), StringIO() jsm.minify(tmpin, tmpout) minified = tmpout.getvalue() - outtxt += unicode(minified or '', 'utf-8').strip('\n') + ';' + if minified: + outtxt += unicode(minified or '', 'utf-8').strip('\n') + ';' if verbose: print "{0}: {1}k".format(f, int(len(minified) / 1024)) diff --git a/frappe/celery_app.py b/frappe/celery_app.py index 715a1b8363..06a7a41b39 100644 --- a/frappe/celery_app.py +++ b/frappe/celery_app.py @@ -17,7 +17,7 @@ SITES_PATH = os.environ.get('SITES_PATH', '.') # defaults DEFAULT_CELERY_BROKER = "redis://localhost" -DEFAULT_CELERY_BACKEND = None +DEFAULT_CELERY_BACKEND = "redis://localhost" DEFAULT_SCHEDULER_INTERVAL = 300 LONGJOBS_PREFIX = "longjobs@" @@ -41,6 +41,7 @@ def setup_celery(app, conf): app.conf.CELERY_TASK_SERIALIZER = 'json' app.conf.CELERY_ACCEPT_CONTENT = ['json'] app.conf.CELERY_TIMEZONE = 'UTC' + app.conf.CELERY_RESULT_SERIALIZER = 'json' if conf.celery_queue_per_site: app.conf.CELERY_ROUTES = (SiteRouter(),) diff --git a/frappe/core/doctype/async_task/__init__.py b/frappe/core/doctype/async_task/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/frappe/core/doctype/async_task/async_task.json b/frappe/core/doctype/async_task/async_task.json new file mode 100644 index 0000000000..7563e70316 --- /dev/null +++ b/frappe/core/doctype/async_task/async_task.json @@ -0,0 +1,142 @@ +{ + "allow_copy": 0, + "allow_import": 0, + "allow_rename": 0, + "autoname": "field:celery_task_id", + "creation": "2015-07-03 11:28:03.496346", + "custom": 0, + "docstatus": 0, + "doctype": "DocType", + "document_type": "Transaction", + "fields": [ + { + "allow_on_submit": 0, + "fieldname": "celery_task_id", + "fieldtype": "Data", + "hidden": 0, + "ignore_user_permissions": 0, + "in_filter": 0, + "in_list_view": 1, + "label": "Celery Task ID", + "no_copy": 0, + "permlevel": 0, + "precision": "", + "print_hide": 0, + "read_only": 0, + "report_hide": 0, + "reqd": 0, + "search_index": 0, + "set_only_once": 0, + "unique": 0 + }, + { + "allow_on_submit": 0, + "fieldname": "status", + "fieldtype": "Select", + "hidden": 0, + "ignore_user_permissions": 0, + "in_filter": 0, + "in_list_view": 0, + "label": "Status", + "no_copy": 0, + "options": "\nQueued\nRunning\nFinished\nFailed\n", + "permlevel": 0, + "precision": "", + "print_hide": 0, + "read_only": 0, + "report_hide": 0, + "reqd": 0, + "search_index": 0, + "set_only_once": 0, + "unique": 0 + }, + { + "allow_on_submit": 0, + "fieldname": "stdout", + "fieldtype": "Long Text", + "hidden": 0, + "ignore_user_permissions": 0, + "in_filter": 0, + "in_list_view": 0, + "label": "stdout", + "no_copy": 0, + "permlevel": 0, + "precision": "", + "print_hide": 0, + "read_only": 1, + "report_hide": 0, + "reqd": 0, + "search_index": 0, + "set_only_once": 0, + "unique": 0 + }, + { + "allow_on_submit": 0, + "fieldname": "stderr", + "fieldtype": "Long Text", + "hidden": 0, + "ignore_user_permissions": 0, + "in_filter": 0, + "in_list_view": 0, + "label": "stderr", + "no_copy": 0, + "permlevel": 0, + "precision": "", + "print_hide": 0, + "read_only": 1, + "report_hide": 0, + "reqd": 0, + "search_index": 0, + "set_only_once": 0, + "unique": 0 + }, + { + "fieldname": "reference_doctype", + "fieldtype": "Link", + "label": "Reference DocType", + "options": "DocType", + "permlevel": 0, + "precision": "" + }, + { + "fieldname": "reference_name", + "fieldtype": "Dynamic Link", + "label": "Reference Doc", + "options": "reference_doctype", + "permlevel": 0, + "precision": "" + } + ], + "hide_heading": 0, + "hide_toolbar": 0, + "in_create": 0, + "in_dialog": 0, + "is_submittable": 0, + "issingle": 0, + "istable": 0, + "modified": "2015-07-04 14:33:26.791024", + "modified_by": "Administrator", + "module": "Core", + "name": "Async Task", + "name_case": "", + "owner": "Administrator", + "permissions": [ + { + "create": 1, + "delete": 1, + "email": 1, + "export": 1, + "permlevel": 0, + "print": 1, + "read": 1, + "report": 1, + "role": "System Manager", + "share": 1, + "write": 1 + } + ], + "read_only": 0, + "read_only_onload": 0, + "sort_field": "modified", + "sort_order": "DESC" +} \ No newline at end of file diff --git a/frappe/core/doctype/async_task/async_task.py b/frappe/core/doctype/async_task/async_task.py new file mode 100644 index 0000000000..85d043524b --- /dev/null +++ b/frappe/core/doctype/async_task/async_task.py @@ -0,0 +1,10 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and contributors +# For license information, please see license.txt + +from __future__ import unicode_literals +import frappe +from frappe.model.document import Document + +class AsyncTask(Document): + pass diff --git a/frappe/core/doctype/async_task/test_async_task.py b/frappe/core/doctype/async_task/test_async_task.py new file mode 100644 index 0000000000..1e885f27f9 --- /dev/null +++ b/frappe/core/doctype/async_task/test_async_task.py @@ -0,0 +1,12 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# See license.txt +from __future__ import unicode_literals + +import frappe +import unittest + +# test_records = frappe.get_test_records('Async Task') + +class TestAsyncTask(unittest.TestCase): + pass diff --git a/frappe/handler.py b/frappe/handler.py index 25e28a063d..05f17b623f 100755 --- a/frappe/handler.py +++ b/frappe/handler.py @@ -70,7 +70,7 @@ def handle(): return build_response("json") -def execute_cmd(cmd): +def execute_cmd(cmd, async=False): """execute a request as python module""" for hook in frappe.get_hooks("override_whitelisted_methods", {}).get(cmd, []): # override using the first hook @@ -78,6 +78,8 @@ def execute_cmd(cmd): break method = get_attr(cmd) + if async: + method = method._f # check if whitelisted if frappe.session['user'] == 'Guest': @@ -103,3 +105,15 @@ def get_attr(cmd): method = globals()[cmd] frappe.log("method:" + cmd) return method + + +@frappe.whitelist() +def get_async_task_status(task_id): + from frappe.celery_app import get_celery + c = get_celery() + a = c.AsyncResult(task_id) + frappe.local.response['response'] = a.result + return { + "state": a.state, + "progress": 0 + } diff --git a/frappe/hooks.py b/frappe/hooks.py index a5c27a9962..bfbefba78f 100644 --- a/frappe/hooks.py +++ b/frappe/hooks.py @@ -147,6 +147,7 @@ scheduler_events = { "frappe.desk.doctype.event.event.send_event_digest", "frappe.sessions.clear_expired_sessions", "frappe.email.doctype.email_alert.email_alert.trigger_daily_alerts", + "frappe.async.remove_old_task_logs", ] } diff --git a/frappe/installer.py b/frappe/installer.py index 4c3c10e538..4f63a1404e 100755 --- a/frappe/installer.py +++ b/frappe/installer.py @@ -214,7 +214,8 @@ def make_site_dirs(): site_private_path = os.path.join(frappe.local.site_path, 'private') for dir_path in ( os.path.join(site_private_path, 'backups'), - os.path.join(site_public_path, 'files')): + os.path.join(site_public_path, 'files'), + os.path.join(site_public_path, 'task-logs')): if not os.path.exists(dir_path): os.makedirs(dir_path) locks_dir = frappe.get_site_path('locks') diff --git a/frappe/public/build.json b/frappe/public/build.json index 7d7cb8fa7a..63ba6d41d8 100644 --- a/frappe/public/build.json +++ b/frappe/public/build.json @@ -15,7 +15,8 @@ "public/js/lib/moment/moment.min.js", "public/js/lib/highlight.pack.js", "public/js/frappe/class.js", - "website/js/website.js" + "website/js/website.js", + "public/js/lib/socket.io.min.js" ], "js/editor.min.js": [ "public/js/lib/jquery/jquery.hotkeys.js", @@ -49,6 +50,7 @@ "public/js/lib/nprogress.js", "public/js/lib/moment/moment-with-locales.min.js", "public/js/lib/moment/moment-timezone-with-data.min.js", + "public/js/lib/socket.io.min.js", "public/js/frappe/provide.js", "public/js/frappe/class.js", @@ -61,6 +63,7 @@ "public/js/frappe/ui/messages.js", "public/js/frappe/request.js", + "public/js/frappe/socket.js", "public/js/frappe/router.js", "public/js/frappe/defaults.js", "public/js/lib/microtemplate.js", diff --git a/frappe/public/js/frappe/request.js b/frappe/public/js/frappe/request.js index 5e200fc442..b437662c52 100644 --- a/frappe/public/js/frappe/request.js +++ b/frappe/public/js/frappe/request.js @@ -28,10 +28,21 @@ frappe.call = function(opts) { args.cmd = opts.method; } + var callback = function(data, xhr) { + if(data.task_id) { + // async call, subscribe + frappe.socket.subscribe(data.task_id, opts); + } + else { + // ajax + return opts.callback(data, xhr); + } + } + return frappe.request.call({ type: opts.type || "POST", args: args, - success: opts.callback, + success: callback, error: opts.error, always: opts.always, btn: opts.btn, diff --git a/frappe/public/js/frappe/socket.js b/frappe/public/js/frappe/socket.js new file mode 100644 index 0000000000..c264292788 --- /dev/null +++ b/frappe/public/js/frappe/socket.js @@ -0,0 +1,60 @@ +frappe.socket = { + open_tasks: {}, + init: function() { + frappe.socket.socket = io.connect('http://' + document.domain + ':' + 3000); + frappe.socket.socket.on('msgprint', function(message) { + frappe.msgprint(message) + }); + + frappe.socket.setup_listeners(); + frappe.socket.setup_reconnect(); + }, + subscribe: function(task_id, opts) { + frappe.socket.socket.emit('task_subscribe', task_id); + frappe.socket.socket.emit('progress_subscribe', task_id); + + frappe.socket.open_tasks[task_id] = opts; + }, + setup_listeners: function() { + frappe.socket.socket.on('task_status_change', function(data) { + if(data.status==="Running") { + frappe.socket.process_response(data, "running"); + } else { + // failed or finished + frappe.socket.process_response(data, "callback"); + // delete frappe.socket.open_tasks[data.task_id]; + } + }); + frappe.socket.socket.on('task_progress', function(data) { + frappe.socket.process_response(data, "progress"); + }); + + }, + setup_reconnect: function() { + // subscribe again to open_tasks + frappe.socket.socket.on("connect", function() { + $.each(frappe.socket.open_tasks, function(task_id, opts) { + frappe.socket.subscribe(task_id, opts); + }); + }); + }, + process_response: function(data, method) { + if(!data) { + return; + } + if(data) { + var opts = frappe.socket.open_tasks[data.task_id]; + if(opts[method]) opts[method](data.message); + } + if(opts.always) { + opts.always(data.message); + } + if(data.status_code && status_code > 400 && opts.error) { + opts.error(data.message); + return; + } + + } +} + +$(frappe.socket.init); diff --git a/frappe/public/js/lib/socket.io.min.js b/frappe/public/js/lib/socket.io.min.js new file mode 100644 index 0000000000..443b5d51b4 --- /dev/null +++ b/frappe/public/js/lib/socket.io.min.js @@ -0,0 +1,7000 @@ +!function(e){if("object"==typeof exports&&"undefined"!=typeof module)module.exports=e();else if("function"==typeof define&&define.amd)define([],e);else{var f;"undefined"!=typeof window?f=window:"undefined"!=typeof global?f=global:"undefined"!=typeof self&&(f=self),f.io=e()}}(function(){var define,module,exports;return (function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);throw new Error("Cannot find module '"+o+"'")}var f=n[o]={exports:{}};t[o][0].call(f.exports,function(e){var n=t[o][1][e];return s(n?n:e)},f,f.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o 0 && !this.encoding) { + var pack = this.packetBuffer.shift(); + this.packet(pack); + } +}; + +/** + * Clean up transport subscriptions and packet buffer. + * + * @api private + */ + +Manager.prototype.cleanup = function(){ + var sub; + while (sub = this.subs.shift()) sub.destroy(); + + this.packetBuffer = []; + this.encoding = false; + + this.decoder.destroy(); +}; + +/** + * Close the current socket. + * + * @api private + */ + +Manager.prototype.close = +Manager.prototype.disconnect = function(){ + this.skipReconnect = true; + this.backoff.reset(); + this.readyState = 'closed'; + this.engine && this.engine.close(); +}; + +/** + * Called upon engine close. + * + * @api private + */ + +Manager.prototype.onclose = function(reason){ + debug('close'); + this.cleanup(); + this.backoff.reset(); + this.readyState = 'closed'; + this.emit('close', reason); + if (this._reconnection && !this.skipReconnect) { + this.reconnect(); + } +}; + +/** + * Attempt a reconnection. + * + * @api private + */ + +Manager.prototype.reconnect = function(){ + if (this.reconnecting || this.skipReconnect) return this; + + var self = this; + + if (this.backoff.attempts >= this._reconnectionAttempts) { + debug('reconnect failed'); + this.backoff.reset(); + this.emitAll('reconnect_failed'); + this.reconnecting = false; + } else { + var delay = this.backoff.duration(); + debug('will wait %dms before reconnect attempt', delay); + + this.reconnecting = true; + var timer = setTimeout(function(){ + if (self.skipReconnect) return; + + debug('attempting reconnect'); + self.emitAll('reconnect_attempt', self.backoff.attempts); + self.emitAll('reconnecting', self.backoff.attempts); + + // check again for the case socket closed in above events + if (self.skipReconnect) return; + + self.open(function(err){ + if (err) { + debug('reconnect attempt error'); + self.reconnecting = false; + self.reconnect(); + self.emitAll('reconnect_error', err.data); + } else { + debug('reconnect success'); + self.onreconnect(); + } + }); + }, delay); + + this.subs.push({ + destroy: function(){ + clearTimeout(timer); + } + }); + } +}; + +/** + * Called upon successful reconnect. + * + * @api private + */ + +Manager.prototype.onreconnect = function(){ + var attempt = this.backoff.attempts; + this.reconnecting = false; + this.backoff.reset(); + this.updateSocketIds(); + this.emitAll('reconnect', attempt); +}; + +},{"./on":4,"./socket":5,"./url":6,"backo2":7,"component-bind":8,"component-emitter":9,"debug":10,"engine.io-client":11,"indexof":42,"object-component":43,"socket.io-parser":46}],4:[function(_dereq_,module,exports){ + +/** + * Module exports. + */ + +module.exports = on; + +/** + * Helper for subscriptions. + * + * @param {Object|EventEmitter} obj with `Emitter` mixin or `EventEmitter` + * @param {String} event name + * @param {Function} callback + * @api public + */ + +function on(obj, ev, fn) { + obj.on(ev, fn); + return { + destroy: function(){ + obj.removeListener(ev, fn); + } + }; +} + +},{}],5:[function(_dereq_,module,exports){ + +/** + * Module dependencies. + */ + +var parser = _dereq_('socket.io-parser'); +var Emitter = _dereq_('component-emitter'); +var toArray = _dereq_('to-array'); +var on = _dereq_('./on'); +var bind = _dereq_('component-bind'); +var debug = _dereq_('debug')('socket.io-client:socket'); +var hasBin = _dereq_('has-binary'); + +/** + * Module exports. + */ + +module.exports = exports = Socket; + +/** + * Internal events (blacklisted). + * These events can't be emitted by the user. + * + * @api private + */ + +var events = { + connect: 1, + connect_error: 1, + connect_timeout: 1, + disconnect: 1, + error: 1, + reconnect: 1, + reconnect_attempt: 1, + reconnect_failed: 1, + reconnect_error: 1, + reconnecting: 1 +}; + +/** + * Shortcut to `Emitter#emit`. + */ + +var emit = Emitter.prototype.emit; + +/** + * `Socket` constructor. + * + * @api public + */ + +function Socket(io, nsp){ + this.io = io; + this.nsp = nsp; + this.json = this; // compat + this.ids = 0; + this.acks = {}; + if (this.io.autoConnect) this.open(); + this.receiveBuffer = []; + this.sendBuffer = []; + this.connected = false; + this.disconnected = true; +} + +/** + * Mix in `Emitter`. + */ + +Emitter(Socket.prototype); + +/** + * Subscribe to open, close and packet events + * + * @api private + */ + +Socket.prototype.subEvents = function() { + if (this.subs) return; + + var io = this.io; + this.subs = [ + on(io, 'open', bind(this, 'onopen')), + on(io, 'packet', bind(this, 'onpacket')), + on(io, 'close', bind(this, 'onclose')) + ]; +}; + +/** + * "Opens" the socket. + * + * @api public + */ + +Socket.prototype.open = +Socket.prototype.connect = function(){ + if (this.connected) return this; + + this.subEvents(); + this.io.open(); // ensure open + if ('open' == this.io.readyState) this.onopen(); + return this; +}; + +/** + * Sends a `message` event. + * + * @return {Socket} self + * @api public + */ + +Socket.prototype.send = function(){ + var args = toArray(arguments); + args.unshift('message'); + this.emit.apply(this, args); + return this; +}; + +/** + * Override `emit`. + * If the event is in `events`, it's emitted normally. + * + * @param {String} event name + * @return {Socket} self + * @api public + */ + +Socket.prototype.emit = function(ev){ + if (events.hasOwnProperty(ev)) { + emit.apply(this, arguments); + return this; + } + + var args = toArray(arguments); + var parserType = parser.EVENT; // default + if (hasBin(args)) { parserType = parser.BINARY_EVENT; } // binary + var packet = { type: parserType, data: args }; + + // event ack callback + if ('function' == typeof args[args.length - 1]) { + debug('emitting packet with ack id %d', this.ids); + this.acks[this.ids] = args.pop(); + packet.id = this.ids++; + } + + if (this.connected) { + this.packet(packet); + } else { + this.sendBuffer.push(packet); + } + + return this; +}; + +/** + * Sends a packet. + * + * @param {Object} packet + * @api private + */ + +Socket.prototype.packet = function(packet){ + packet.nsp = this.nsp; + this.io.packet(packet); +}; + +/** + * Called upon engine `open`. + * + * @api private + */ + +Socket.prototype.onopen = function(){ + debug('transport is open - connecting'); + + // write connect packet if necessary + if ('/' != this.nsp) { + this.packet({ type: parser.CONNECT }); + } +}; + +/** + * Called upon engine `close`. + * + * @param {String} reason + * @api private + */ + +Socket.prototype.onclose = function(reason){ + debug('close (%s)', reason); + this.connected = false; + this.disconnected = true; + delete this.id; + this.emit('disconnect', reason); +}; + +/** + * Called with socket packet. + * + * @param {Object} packet + * @api private + */ + +Socket.prototype.onpacket = function(packet){ + if (packet.nsp != this.nsp) return; + + switch (packet.type) { + case parser.CONNECT: + this.onconnect(); + break; + + case parser.EVENT: + this.onevent(packet); + break; + + case parser.BINARY_EVENT: + this.onevent(packet); + break; + + case parser.ACK: + this.onack(packet); + break; + + case parser.BINARY_ACK: + this.onack(packet); + break; + + case parser.DISCONNECT: + this.ondisconnect(); + break; + + case parser.ERROR: + this.emit('error', packet.data); + break; + } +}; + +/** + * Called upon a server event. + * + * @param {Object} packet + * @api private + */ + +Socket.prototype.onevent = function(packet){ + var args = packet.data || []; + debug('emitting event %j', args); + + if (null != packet.id) { + debug('attaching ack callback to event'); + args.push(this.ack(packet.id)); + } + + if (this.connected) { + emit.apply(this, args); + } else { + this.receiveBuffer.push(args); + } +}; + +/** + * Produces an ack callback to emit with an event. + * + * @api private + */ + +Socket.prototype.ack = function(id){ + var self = this; + var sent = false; + return function(){ + // prevent double callbacks + if (sent) return; + sent = true; + var args = toArray(arguments); + debug('sending ack %j', args); + + var type = hasBin(args) ? parser.BINARY_ACK : parser.ACK; + self.packet({ + type: type, + id: id, + data: args + }); + }; +}; + +/** + * Called upon a server acknowlegement. + * + * @param {Object} packet + * @api private + */ + +Socket.prototype.onack = function(packet){ + debug('calling ack %s with %j', packet.id, packet.data); + var fn = this.acks[packet.id]; + fn.apply(this, packet.data); + delete this.acks[packet.id]; +}; + +/** + * Called upon server connect. + * + * @api private + */ + +Socket.prototype.onconnect = function(){ + this.connected = true; + this.disconnected = false; + this.emit('connect'); + this.emitBuffered(); +}; + +/** + * Emit buffered events (received and emitted). + * + * @api private + */ + +Socket.prototype.emitBuffered = function(){ + var i; + for (i = 0; i < this.receiveBuffer.length; i++) { + emit.apply(this, this.receiveBuffer[i]); + } + this.receiveBuffer = []; + + for (i = 0; i < this.sendBuffer.length; i++) { + this.packet(this.sendBuffer[i]); + } + this.sendBuffer = []; +}; + +/** + * Called upon server disconnect. + * + * @api private + */ + +Socket.prototype.ondisconnect = function(){ + debug('server disconnect (%s)', this.nsp); + this.destroy(); + this.onclose('io server disconnect'); +}; + +/** + * Called upon forced client/server side disconnections, + * this method ensures the manager stops tracking us and + * that reconnections don't get triggered for this. + * + * @api private. + */ + +Socket.prototype.destroy = function(){ + if (this.subs) { + // clean subscriptions to avoid reconnections + for (var i = 0; i < this.subs.length; i++) { + this.subs[i].destroy(); + } + this.subs = null; + } + + this.io.destroy(this); +}; + +/** + * Disconnects the socket manually. + * + * @return {Socket} self + * @api public + */ + +Socket.prototype.close = +Socket.prototype.disconnect = function(){ + if (this.connected) { + debug('performing disconnect (%s)', this.nsp); + this.packet({ type: parser.DISCONNECT }); + } + + // remove socket from pool + this.destroy(); + + if (this.connected) { + // fire events + this.onclose('io client disconnect'); + } + return this; +}; + +},{"./on":4,"component-bind":8,"component-emitter":9,"debug":10,"has-binary":38,"socket.io-parser":46,"to-array":50}],6:[function(_dereq_,module,exports){ +(function (global){ + +/** + * Module dependencies. + */ + +var parseuri = _dereq_('parseuri'); +var debug = _dereq_('debug')('socket.io-client:url'); + +/** + * Module exports. + */ + +module.exports = url; + +/** + * URL parser. + * + * @param {String} url + * @param {Object} An object meant to mimic window.location. + * Defaults to window.location. + * @api public + */ + +function url(uri, loc){ + var obj = uri; + + // default to window.location + var loc = loc || global.location; + if (null == uri) uri = loc.protocol + '//' + loc.host; + + // relative path support + if ('string' == typeof uri) { + if ('/' == uri.charAt(0)) { + if ('/' == uri.charAt(1)) { + uri = loc.protocol + uri; + } else { + uri = loc.hostname + uri; + } + } + + if (!/^(https?|wss?):\/\//.test(uri)) { + debug('protocol-less url %s', uri); + if ('undefined' != typeof loc) { + uri = loc.protocol + '//' + uri; + } else { + uri = 'https://' + uri; + } + } + + // parse + debug('parse %s', uri); + obj = parseuri(uri); + } + + // make sure we treat `localhost:80` and `localhost` equally + if (!obj.port) { + if (/^(http|ws)$/.test(obj.protocol)) { + obj.port = '80'; + } + else if (/^(http|ws)s$/.test(obj.protocol)) { + obj.port = '443'; + } + } + + obj.path = obj.path || '/'; + + // define unique id + obj.id = obj.protocol + '://' + obj.host + ':' + obj.port; + // define href + obj.href = obj.protocol + '://' + obj.host + (loc && loc.port == obj.port ? '' : (':' + obj.port)); + + return obj; +} + +}).call(this,typeof self !== "undefined" ? self : typeof window !== "undefined" ? window : {}) +},{"debug":10,"parseuri":44}],7:[function(_dereq_,module,exports){ + +/** + * Expose `Backoff`. + */ + +module.exports = Backoff; + +/** + * Initialize backoff timer with `opts`. + * + * - `min` initial timeout in milliseconds [100] + * - `max` max timeout [10000] + * - `jitter` [0] + * - `factor` [2] + * + * @param {Object} opts + * @api public + */ + +function Backoff(opts) { + opts = opts || {}; + this.ms = opts.min || 100; + this.max = opts.max || 10000; + this.factor = opts.factor || 2; + this.jitter = opts.jitter > 0 && opts.jitter <= 1 ? opts.jitter : 0; + this.attempts = 0; +} + +/** + * Return the backoff duration. + * + * @return {Number} + * @api public + */ + +Backoff.prototype.duration = function(){ + var ms = this.ms * Math.pow(this.factor, this.attempts++); + if (this.jitter) { + var rand = Math.random(); + var deviation = Math.floor(rand * this.jitter * ms); + ms = (Math.floor(rand * 10) & 1) == 0 ? ms - deviation : ms + deviation; + } + return Math.min(ms, this.max) | 0; +}; + +/** + * Reset the number of attempts. + * + * @api public + */ + +Backoff.prototype.reset = function(){ + this.attempts = 0; +}; + +/** + * Set the minimum duration + * + * @api public + */ + +Backoff.prototype.setMin = function(min){ + this.ms = min; +}; + +/** + * Set the maximum duration + * + * @api public + */ + +Backoff.prototype.setMax = function(max){ + this.max = max; +}; + +/** + * Set the jitter + * + * @api public + */ + +Backoff.prototype.setJitter = function(jitter){ + this.jitter = jitter; +}; + + +},{}],8:[function(_dereq_,module,exports){ +/** + * Slice reference. + */ + +var slice = [].slice; + +/** + * Bind `obj` to `fn`. + * + * @param {Object} obj + * @param {Function|String} fn or string + * @return {Function} + * @api public + */ + +module.exports = function(obj, fn){ + if ('string' == typeof fn) fn = obj[fn]; + if ('function' != typeof fn) throw new Error('bind() requires a function'); + var args = slice.call(arguments, 2); + return function(){ + return fn.apply(obj, args.concat(slice.call(arguments))); + } +}; + +},{}],9:[function(_dereq_,module,exports){ + +/** + * Expose `Emitter`. + */ + +module.exports = Emitter; + +/** + * Initialize a new `Emitter`. + * + * @api public + */ + +function Emitter(obj) { + if (obj) return mixin(obj); +}; + +/** + * Mixin the emitter properties. + * + * @param {Object} obj + * @return {Object} + * @api private + */ + +function mixin(obj) { + for (var key in Emitter.prototype) { + obj[key] = Emitter.prototype[key]; + } + return obj; +} + +/** + * Listen on the given `event` with `fn`. + * + * @param {String} event + * @param {Function} fn + * @return {Emitter} + * @api public + */ + +Emitter.prototype.on = +Emitter.prototype.addEventListener = function(event, fn){ + this._callbacks = this._callbacks || {}; + (this._callbacks[event] = this._callbacks[event] || []) + .push(fn); + return this; +}; + +/** + * Adds an `event` listener that will be invoked a single + * time then automatically removed. + * + * @param {String} event + * @param {Function} fn + * @return {Emitter} + * @api public + */ + +Emitter.prototype.once = function(event, fn){ + var self = this; + this._callbacks = this._callbacks || {}; + + function on() { + self.off(event, on); + fn.apply(this, arguments); + } + + on.fn = fn; + this.on(event, on); + return this; +}; + +/** + * Remove the given callback for `event` or all + * registered callbacks. + * + * @param {String} event + * @param {Function} fn + * @return {Emitter} + * @api public + */ + +Emitter.prototype.off = +Emitter.prototype.removeListener = +Emitter.prototype.removeAllListeners = +Emitter.prototype.removeEventListener = function(event, fn){ + this._callbacks = this._callbacks || {}; + + // all + if (0 == arguments.length) { + this._callbacks = {}; + return this; + } + + // specific event + var callbacks = this._callbacks[event]; + if (!callbacks) return this; + + // remove all handlers + if (1 == arguments.length) { + delete this._callbacks[event]; + return this; + } + + // remove specific handler + var cb; + for (var i = 0; i < callbacks.length; i++) { + cb = callbacks[i]; + if (cb === fn || cb.fn === fn) { + callbacks.splice(i, 1); + break; + } + } + return this; +}; + +/** + * Emit `event` with the given args. + * + * @param {String} event + * @param {Mixed} ... + * @return {Emitter} + */ + +Emitter.prototype.emit = function(event){ + this._callbacks = this._callbacks || {}; + var args = [].slice.call(arguments, 1) + , callbacks = this._callbacks[event]; + + if (callbacks) { + callbacks = callbacks.slice(0); + for (var i = 0, len = callbacks.length; i < len; ++i) { + callbacks[i].apply(this, args); + } + } + + return this; +}; + +/** + * Return array of callbacks for `event`. + * + * @param {String} event + * @return {Array} + * @api public + */ + +Emitter.prototype.listeners = function(event){ + this._callbacks = this._callbacks || {}; + return this._callbacks[event] || []; +}; + +/** + * Check if this emitter has `event` handlers. + * + * @param {String} event + * @return {Boolean} + * @api public + */ + +Emitter.prototype.hasListeners = function(event){ + return !! this.listeners(event).length; +}; + +},{}],10:[function(_dereq_,module,exports){ + +/** + * Expose `debug()` as the module. + */ + +module.exports = debug; + +/** + * Create a debugger with the given `name`. + * + * @param {String} name + * @return {Type} + * @api public + */ + +function debug(name) { + if (!debug.enabled(name)) return function(){}; + + return function(fmt){ + fmt = coerce(fmt); + + var curr = new Date; + var ms = curr - (debug[name] || curr); + debug[name] = curr; + + fmt = name + + ' ' + + fmt + + ' +' + debug.humanize(ms); + + // This hackery is required for IE8 + // where `console.log` doesn't have 'apply' + window.console + && console.log + && Function.prototype.apply.call(console.log, console, arguments); + } +} + +/** + * The currently active debug mode names. + */ + +debug.names = []; +debug.skips = []; + +/** + * Enables a debug mode by name. This can include modes + * separated by a colon and wildcards. + * + * @param {String} name + * @api public + */ + +debug.enable = function(name) { + try { + localStorage.debug = name; + } catch(e){} + + var split = (name || '').split(/[\s,]+/) + , len = split.length; + + for (var i = 0; i < len; i++) { + name = split[i].replace('*', '.*?'); + if (name[0] === '-') { + debug.skips.push(new RegExp('^' + name.substr(1) + '$')); + } + else { + debug.names.push(new RegExp('^' + name + '$')); + } + } +}; + +/** + * Disable debug output. + * + * @api public + */ + +debug.disable = function(){ + debug.enable(''); +}; + +/** + * Humanize the given `ms`. + * + * @param {Number} m + * @return {String} + * @api private + */ + +debug.humanize = function(ms) { + var sec = 1000 + , min = 60 * 1000 + , hour = 60 * min; + + if (ms >= hour) return (ms / hour).toFixed(1) + 'h'; + if (ms >= min) return (ms / min).toFixed(1) + 'm'; + if (ms >= sec) return (ms / sec | 0) + 's'; + return ms + 'ms'; +}; + +/** + * Returns true if the given mode name is enabled, false otherwise. + * + * @param {String} name + * @return {Boolean} + * @api public + */ + +debug.enabled = function(name) { + for (var i = 0, len = debug.skips.length; i < len; i++) { + if (debug.skips[i].test(name)) { + return false; + } + } + for (var i = 0, len = debug.names.length; i < len; i++) { + if (debug.names[i].test(name)) { + return true; + } + } + return false; +}; + +/** + * Coerce `val`. + */ + +function coerce(val) { + if (val instanceof Error) return val.stack || val.message; + return val; +} + +// persist + +try { + if (window.localStorage) debug.enable(localStorage.debug); +} catch(e){} + +},{}],11:[function(_dereq_,module,exports){ + +module.exports = _dereq_('./lib/'); + +},{"./lib/":12}],12:[function(_dereq_,module,exports){ + +module.exports = _dereq_('./socket'); + +/** + * Exports parser + * + * @api public + * + */ +module.exports.parser = _dereq_('engine.io-parser'); + +},{"./socket":13,"engine.io-parser":25}],13:[function(_dereq_,module,exports){ +(function (global){ +/** + * Module dependencies. + */ + +var transports = _dereq_('./transports'); +var Emitter = _dereq_('component-emitter'); +var debug = _dereq_('debug')('engine.io-client:socket'); +var index = _dereq_('indexof'); +var parser = _dereq_('engine.io-parser'); +var parseuri = _dereq_('parseuri'); +var parsejson = _dereq_('parsejson'); +var parseqs = _dereq_('parseqs'); + +/** + * Module exports. + */ + +module.exports = Socket; + +/** + * Noop function. + * + * @api private + */ + +function noop(){} + +/** + * Socket constructor. + * + * @param {String|Object} uri or options + * @param {Object} options + * @api public + */ + +function Socket(uri, opts){ + if (!(this instanceof Socket)) return new Socket(uri, opts); + + opts = opts || {}; + + if (uri && 'object' == typeof uri) { + opts = uri; + uri = null; + } + + if (uri) { + uri = parseuri(uri); + opts.host = uri.host; + opts.secure = uri.protocol == 'https' || uri.protocol == 'wss'; + opts.port = uri.port; + if (uri.query) opts.query = uri.query; + } + + this.secure = null != opts.secure ? opts.secure : + (global.location && 'https:' == location.protocol); + + if (opts.host) { + var pieces = opts.host.split(':'); + opts.hostname = pieces.shift(); + if (pieces.length) { + opts.port = pieces.pop(); + } else if (!opts.port) { + // if no port is specified manually, use the protocol default + opts.port = this.secure ? '443' : '80'; + } + } + + this.agent = opts.agent || false; + this.hostname = opts.hostname || + (global.location ? location.hostname : 'localhost'); + this.port = opts.port || (global.location && location.port ? + location.port : + (this.secure ? 443 : 80)); + this.query = opts.query || {}; + if ('string' == typeof this.query) this.query = parseqs.decode(this.query); + this.upgrade = false !== opts.upgrade; + this.path = (opts.path || '/engine.io').replace(/\/$/, '') + '/'; + this.forceJSONP = !!opts.forceJSONP; + this.jsonp = false !== opts.jsonp; + this.forceBase64 = !!opts.forceBase64; + this.enablesXDR = !!opts.enablesXDR; + this.timestampParam = opts.timestampParam || 't'; + this.timestampRequests = opts.timestampRequests; + this.transports = opts.transports || ['polling', 'websocket']; + this.readyState = ''; + this.writeBuffer = []; + this.callbackBuffer = []; + this.policyPort = opts.policyPort || 843; + this.rememberUpgrade = opts.rememberUpgrade || false; + this.binaryType = null; + this.onlyBinaryUpgrades = opts.onlyBinaryUpgrades; + + // SSL options for Node.js client + this.pfx = opts.pfx || null; + this.key = opts.key || null; + this.passphrase = opts.passphrase || null; + this.cert = opts.cert || null; + this.ca = opts.ca || null; + this.ciphers = opts.ciphers || null; + this.rejectUnauthorized = opts.rejectUnauthorized || null; + + this.open(); +} + +Socket.priorWebsocketSuccess = false; + +/** + * Mix in `Emitter`. + */ + +Emitter(Socket.prototype); + +/** + * Protocol version. + * + * @api public + */ + +Socket.protocol = parser.protocol; // this is an int + +/** + * Expose deps for legacy compatibility + * and standalone browser access. + */ + +Socket.Socket = Socket; +Socket.Transport = _dereq_('./transport'); +Socket.transports = _dereq_('./transports'); +Socket.parser = _dereq_('engine.io-parser'); + +/** + * Creates transport of the given type. + * + * @param {String} transport name + * @return {Transport} + * @api private + */ + +Socket.prototype.createTransport = function (name) { + debug('creating transport "%s"', name); + var query = clone(this.query); + + // append engine.io protocol identifier + query.EIO = parser.protocol; + + // transport name + query.transport = name; + + // session id if we already have one + if (this.id) query.sid = this.id; + + var transport = new transports[name]({ + agent: this.agent, + hostname: this.hostname, + port: this.port, + secure: this.secure, + path: this.path, + query: query, + forceJSONP: this.forceJSONP, + jsonp: this.jsonp, + forceBase64: this.forceBase64, + enablesXDR: this.enablesXDR, + timestampRequests: this.timestampRequests, + timestampParam: this.timestampParam, + policyPort: this.policyPort, + socket: this, + pfx: this.pfx, + key: this.key, + passphrase: this.passphrase, + cert: this.cert, + ca: this.ca, + ciphers: this.ciphers, + rejectUnauthorized: this.rejectUnauthorized + }); + + return transport; +}; + +function clone (obj) { + var o = {}; + for (var i in obj) { + if (obj.hasOwnProperty(i)) { + o[i] = obj[i]; + } + } + return o; +} + +/** + * Initializes transport to use and starts probe. + * + * @api private + */ +Socket.prototype.open = function () { + var transport; + if (this.rememberUpgrade && Socket.priorWebsocketSuccess && this.transports.indexOf('websocket') != -1) { + transport = 'websocket'; + } else if (0 == this.transports.length) { + // Emit error on next tick so it can be listened to + var self = this; + setTimeout(function() { + self.emit('error', 'No transports available'); + }, 0); + return; + } else { + transport = this.transports[0]; + } + this.readyState = 'opening'; + + // Retry with the next transport if the transport is disabled (jsonp: false) + var transport; + try { + transport = this.createTransport(transport); + } catch (e) { + this.transports.shift(); + this.open(); + return; + } + + transport.open(); + this.setTransport(transport); +}; + +/** + * Sets the current transport. Disables the existing one (if any). + * + * @api private + */ + +Socket.prototype.setTransport = function(transport){ + debug('setting transport %s', transport.name); + var self = this; + + if (this.transport) { + debug('clearing existing transport %s', this.transport.name); + this.transport.removeAllListeners(); + } + + // set up transport + this.transport = transport; + + // set up transport listeners + transport + .on('drain', function(){ + self.onDrain(); + }) + .on('packet', function(packet){ + self.onPacket(packet); + }) + .on('error', function(e){ + self.onError(e); + }) + .on('close', function(){ + self.onClose('transport close'); + }); +}; + +/** + * Probes a transport. + * + * @param {String} transport name + * @api private + */ + +Socket.prototype.probe = function (name) { + debug('probing transport "%s"', name); + var transport = this.createTransport(name, { probe: 1 }) + , failed = false + , self = this; + + Socket.priorWebsocketSuccess = false; + + function onTransportOpen(){ + if (self.onlyBinaryUpgrades) { + var upgradeLosesBinary = !this.supportsBinary && self.transport.supportsBinary; + failed = failed || upgradeLosesBinary; + } + if (failed) return; + + debug('probe transport "%s" opened', name); + transport.send([{ type: 'ping', data: 'probe' }]); + transport.once('packet', function (msg) { + if (failed) return; + if ('pong' == msg.type && 'probe' == msg.data) { + debug('probe transport "%s" pong', name); + self.upgrading = true; + self.emit('upgrading', transport); + if (!transport) return; + Socket.priorWebsocketSuccess = 'websocket' == transport.name; + + debug('pausing current transport "%s"', self.transport.name); + self.transport.pause(function () { + if (failed) return; + if ('closed' == self.readyState) return; + debug('changing transport and sending upgrade packet'); + + cleanup(); + + self.setTransport(transport); + transport.send([{ type: 'upgrade' }]); + self.emit('upgrade', transport); + transport = null; + self.upgrading = false; + self.flush(); + }); + } else { + debug('probe transport "%s" failed', name); + var err = new Error('probe error'); + err.transport = transport.name; + self.emit('upgradeError', err); + } + }); + } + + function freezeTransport() { + if (failed) return; + + // Any callback called by transport should be ignored since now + failed = true; + + cleanup(); + + transport.close(); + transport = null; + } + + //Handle any error that happens while probing + function onerror(err) { + var error = new Error('probe error: ' + err); + error.transport = transport.name; + + freezeTransport(); + + debug('probe transport "%s" failed because of error: %s', name, err); + + self.emit('upgradeError', error); + } + + function onTransportClose(){ + onerror("transport closed"); + } + + //When the socket is closed while we're probing + function onclose(){ + onerror("socket closed"); + } + + //When the socket is upgraded while we're probing + function onupgrade(to){ + if (transport && to.name != transport.name) { + debug('"%s" works - aborting "%s"', to.name, transport.name); + freezeTransport(); + } + } + + //Remove all listeners on the transport and on self + function cleanup(){ + transport.removeListener('open', onTransportOpen); + transport.removeListener('error', onerror); + transport.removeListener('close', onTransportClose); + self.removeListener('close', onclose); + self.removeListener('upgrading', onupgrade); + } + + transport.once('open', onTransportOpen); + transport.once('error', onerror); + transport.once('close', onTransportClose); + + this.once('close', onclose); + this.once('upgrading', onupgrade); + + transport.open(); + +}; + +/** + * Called when connection is deemed open. + * + * @api public + */ + +Socket.prototype.onOpen = function () { + debug('socket open'); + this.readyState = 'open'; + Socket.priorWebsocketSuccess = 'websocket' == this.transport.name; + this.emit('open'); + this.flush(); + + // we check for `readyState` in case an `open` + // listener already closed the socket + if ('open' == this.readyState && this.upgrade && this.transport.pause) { + debug('starting upgrade probes'); + for (var i = 0, l = this.upgrades.length; i < l; i++) { + this.probe(this.upgrades[i]); + } + } +}; + +/** + * Handles a packet. + * + * @api private + */ + +Socket.prototype.onPacket = function (packet) { + if ('opening' == this.readyState || 'open' == this.readyState) { + debug('socket receive: type "%s", data "%s"', packet.type, packet.data); + + this.emit('packet', packet); + + // Socket is live - any packet counts + this.emit('heartbeat'); + + switch (packet.type) { + case 'open': + this.onHandshake(parsejson(packet.data)); + break; + + case 'pong': + this.setPing(); + break; + + case 'error': + var err = new Error('server error'); + err.code = packet.data; + this.emit('error', err); + break; + + case 'message': + this.emit('data', packet.data); + this.emit('message', packet.data); + break; + } + } else { + debug('packet received with socket readyState "%s"', this.readyState); + } +}; + +/** + * Called upon handshake completion. + * + * @param {Object} handshake obj + * @api private + */ + +Socket.prototype.onHandshake = function (data) { + this.emit('handshake', data); + this.id = data.sid; + this.transport.query.sid = data.sid; + this.upgrades = this.filterUpgrades(data.upgrades); + this.pingInterval = data.pingInterval; + this.pingTimeout = data.pingTimeout; + this.onOpen(); + // In case open handler closes socket + if ('closed' == this.readyState) return; + this.setPing(); + + // Prolong liveness of socket on heartbeat + this.removeListener('heartbeat', this.onHeartbeat); + this.on('heartbeat', this.onHeartbeat); +}; + +/** + * Resets ping timeout. + * + * @api private + */ + +Socket.prototype.onHeartbeat = function (timeout) { + clearTimeout(this.pingTimeoutTimer); + var self = this; + self.pingTimeoutTimer = setTimeout(function () { + if ('closed' == self.readyState) return; + self.onClose('ping timeout'); + }, timeout || (self.pingInterval + self.pingTimeout)); +}; + +/** + * Pings server every `this.pingInterval` and expects response + * within `this.pingTimeout` or closes connection. + * + * @api private + */ + +Socket.prototype.setPing = function () { + var self = this; + clearTimeout(self.pingIntervalTimer); + self.pingIntervalTimer = setTimeout(function () { + debug('writing ping packet - expecting pong within %sms', self.pingTimeout); + self.ping(); + self.onHeartbeat(self.pingTimeout); + }, self.pingInterval); +}; + +/** +* Sends a ping packet. +* +* @api public +*/ + +Socket.prototype.ping = function () { + this.sendPacket('ping'); +}; + +/** + * Called on `drain` event + * + * @api private + */ + +Socket.prototype.onDrain = function() { + for (var i = 0; i < this.prevBufferLen; i++) { + if (this.callbackBuffer[i]) { + this.callbackBuffer[i](); + } + } + + this.writeBuffer.splice(0, this.prevBufferLen); + this.callbackBuffer.splice(0, this.prevBufferLen); + + // setting prevBufferLen = 0 is very important + // for example, when upgrading, upgrade packet is sent over, + // and a nonzero prevBufferLen could cause problems on `drain` + this.prevBufferLen = 0; + + if (this.writeBuffer.length == 0) { + this.emit('drain'); + } else { + this.flush(); + } +}; + +/** + * Flush write buffers. + * + * @api private + */ + +Socket.prototype.flush = function () { + if ('closed' != this.readyState && this.transport.writable && + !this.upgrading && this.writeBuffer.length) { + debug('flushing %d packets in socket', this.writeBuffer.length); + this.transport.send(this.writeBuffer); + // keep track of current length of writeBuffer + // splice writeBuffer and callbackBuffer on `drain` + this.prevBufferLen = this.writeBuffer.length; + this.emit('flush'); + } +}; + +/** + * Sends a message. + * + * @param {String} message. + * @param {Function} callback function. + * @return {Socket} for chaining. + * @api public + */ + +Socket.prototype.write = +Socket.prototype.send = function (msg, fn) { + this.sendPacket('message', msg, fn); + return this; +}; + +/** + * Sends a packet. + * + * @param {String} packet type. + * @param {String} data. + * @param {Function} callback function. + * @api private + */ + +Socket.prototype.sendPacket = function (type, data, fn) { + if ('closing' == this.readyState || 'closed' == this.readyState) { + return; + } + + var packet = { type: type, data: data }; + this.emit('packetCreate', packet); + this.writeBuffer.push(packet); + this.callbackBuffer.push(fn); + this.flush(); +}; + +/** + * Closes the connection. + * + * @api private + */ + +Socket.prototype.close = function () { + if ('opening' == this.readyState || 'open' == this.readyState) { + this.readyState = 'closing'; + + var self = this; + + function close() { + self.onClose('forced close'); + debug('socket closing - telling transport to close'); + self.transport.close(); + } + + function cleanupAndClose() { + self.removeListener('upgrade', cleanupAndClose); + self.removeListener('upgradeError', cleanupAndClose); + close(); + } + + function waitForUpgrade() { + // wait for upgrade to finish since we can't send packets while pausing a transport + self.once('upgrade', cleanupAndClose); + self.once('upgradeError', cleanupAndClose); + } + + if (this.writeBuffer.length) { + this.once('drain', function() { + if (this.upgrading) { + waitForUpgrade(); + } else { + close(); + } + }); + } else if (this.upgrading) { + waitForUpgrade(); + } else { + close(); + } + } + + return this; +}; + +/** + * Called upon transport error + * + * @api private + */ + +Socket.prototype.onError = function (err) { + debug('socket error %j', err); + Socket.priorWebsocketSuccess = false; + this.emit('error', err); + this.onClose('transport error', err); +}; + +/** + * Called upon transport close. + * + * @api private + */ + +Socket.prototype.onClose = function (reason, desc) { + if ('opening' == this.readyState || 'open' == this.readyState || 'closing' == this.readyState) { + debug('socket close with reason: "%s"', reason); + var self = this; + + // clear timers + clearTimeout(this.pingIntervalTimer); + clearTimeout(this.pingTimeoutTimer); + + // clean buffers in next tick, so developers can still + // grab the buffers on `close` event + setTimeout(function() { + self.writeBuffer = []; + self.callbackBuffer = []; + self.prevBufferLen = 0; + }, 0); + + // stop event from firing again for transport + this.transport.removeAllListeners('close'); + + // ensure transport won't stay open + this.transport.close(); + + // ignore further transport communication + this.transport.removeAllListeners(); + + // set ready state + this.readyState = 'closed'; + + // clear session id + this.id = null; + + // emit close event + this.emit('close', reason, desc); + } +}; + +/** + * Filters upgrades, returning only those matching client transports. + * + * @param {Array} server upgrades + * @api private + * + */ + +Socket.prototype.filterUpgrades = function (upgrades) { + var filteredUpgrades = []; + for (var i = 0, j = upgrades.length; i