Merge branch 'dev/clickClient' into devel

src/octoprint/cli/__init__.py

@@ -119,9 +119,10 @@ legacy_options = bulk_options([
 from .server import server_commands
 from .plugins import plugin_commands
 from .dev import dev_commands
+from .client import client_commands
 
 @click.group(name="octoprint", invoke_without_command=True, cls=click.CommandCollection,
-             sources=[server_commands, plugin_commands, dev_commands])
+             sources=[server_commands, plugin_commands, dev_commands, client_commands])
 @standard_options()
 @legacy_options
 @click.version_option(version=octoprint.__version__)

src/octoprint/cli/client.py

@@ -0,0 +1,176 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+import click
+import json
+
+import octoprint_client
+
+from octoprint.cli import pass_octoprint_ctx, bulk_options, standard_options
+from octoprint import init_settings
+
+
+class JsonStringParamType(click.ParamType):
+	name = "json"
+
+	def convert(self, value, param, ctx):
+		try:
+			return json.loads(value)
+		except:
+			self.fail("%s is not a valid json string" % value, param, ctx)
+
+
+@click.group()
+def client_commands():
+	pass
+
+
+@client_commands.group("client", context_settings=dict(ignore_unknown_options=True))
+@click.option("--host", "-h", type=click.STRING)
+@click.option("--port", "-p", type=click.INT)
+@click.option("--httpuser", type=click.STRING)
+@click.option("--httppass", type=click.STRING)
+@click.option("--https", is_flag=True)
+@click.option("--prefix", type=click.STRING)
+@pass_octoprint_ctx
+def client(obj, host, port, httpuser, httppass, https, prefix):
+	"""Basic API client."""
+	obj.settings = init_settings(obj.basedir, obj.configfile)
+	octoprint_client.init_client(obj.settings, https=https, httpuser=httpuser, httppass=httppass, host=host, port=port, prefix=prefix)
+
+
+def log_response(response, status_code=True, body=True, headers=False):
+	if status_code:
+		click.echo("Status Code: {}".format(response.status_code))
+	if headers:
+		for header, value in response.headers.items():
+			click.echo("{}: {}".format(header, value))
+		click.echo()
+	if body:
+		click.echo(response.text)
+
+
+@client.command("get")
+@click.argument("path")
+def get(path):
+	"""Performs a GET request against the specified server path."""
+	r = octoprint_client.get(path)
+	log_response(r)
+
+
+@client.command("post_json")
+@click.argument("path")
+@click.argument("data", type=JsonStringParamType())
+def post_json(path, data):
+	"""POSTs JSON data to the specified server path."""
+	r = octoprint_client.post_json(path, data)
+	log_response(r)
+
+
+@client.command("patch_json")
+@click.argument("path")
+@click.argument("data", type=JsonStringParamType())
+def patch_json(path, data):
+	"""PATCHes JSON data to the specified server path."""
+	r = octoprint_client.patch(path, data, encoding="json")
+	log_response(r)
+
+
+@client.command("post_from_file")
+@click.argument("path")
+@click.argument("file_path", type=click.Path(exists=True, dir_okay=False, resolve_path=True))
+@click.option("--json", is_flag=True)
+@click.option("--yaml", is_flag=True)
+def post_from_file(path, file_path, json_flag, yaml_flag):
+	"""POSTs JSON data to the specified server path."""
+	if json_flag or yaml_flag:
+		if json_flag:
+			with open(file_path, "rb") as fp:
+				data = json.load(fp)
+		else:
+			import yaml
+			with open(file_path, "rb") as fp:
+				data = yaml.safe_load(fp)
+
+		r = octoprint_client.post_json(path, data)
+	else:
+		with open(file_path, "rb") as fp:
+			data = fp.read()
+
+		r = octoprint_client.post(path, data)
+
+	log_response(r)
+
+
+@client.command("command")
+@click.argument("path")
+@click.argument("command")
+@click.option("--str", "-s", "str_params", multiple=True, nargs=2, type=click.Tuple([unicode, unicode]))
+@click.option("--int", "-i", "int_params", multiple=True, nargs=2, type=click.Tuple([unicode, int]))
+@click.option("--float", "-f", "float_params", multiple=True, nargs=2, type=click.Tuple([unicode, float]))
+@click.option("--bool", "-b", "bool_params", multiple=True, nargs=2, type=click.Tuple([unicode, bool]))
+def command(path, command, str_params, int_params, float_params, bool_params):
+	"""Sends a JSON command to the specified server path."""
+	data = dict()
+	params = str_params + int_params + float_params + bool_params
+	for param in params:
+		data[param[0]] = param[1]
+	r = octoprint_client.post_command(path, command, additional=data)
+	log_response(r, body=False)
+
+
+@client.command("upload")
+@click.argument("path")
+@click.argument("file_path", type=click.Path(exists=True, dir_okay=False, resolve_path=True))
+@click.option("--parameter", "-P", "params", multiple=True, nargs=2, type=click.Tuple([unicode, unicode]))
+@click.option("--file-name", type=click.STRING)
+@click.option("--content-type", type=click.STRING)
+def upload(path, file_path, params, file_name, content_type):
+	"""Uploads the specified file to the specified server path."""
+	data = dict()
+	for param in params:
+		data[param[0]] = param[1]
+
+	r = octoprint_client.upload(path, file_path, additional=data, file_name=file_name, content_type=content_type)
+	log_response(r)
+
+
+@client.command("delete")
+@click.argument("path")
+def delete(path):
+	"""Sends a DELETE request to the specified server path."""
+	r = octoprint_client.delete(path)
+	log_response(r)
+
+
+@client.command("listen")
+def listen():
+	def on_connect(ws):
+		click.echo(">>> Connected!")
+
+	def on_close(ws):
+		click.echo(">>> Connection closed!")
+
+	def on_error(ws, error):
+		click.echo("!!! Error: {}".format(error))
+
+	def on_heartbeat(ws):
+		click.echo("<3")
+
+	def on_message(ws, message_type, message_payload):
+		click.echo("Message: {}, Payload: {}".format(message_type, json.dumps(message_payload)))
+
+	socket = octoprint_client.connect_socket(on_connect=on_connect,
+	                                         on_close=on_close,
+	                                         on_error=on_error,
+	                                         on_heartbeat=on_heartbeat,
+	                                         on_message=on_message)
+
+	click.echo(">>> Waiting for client to exit")
+	try:
+		socket.wait()
+	finally:
+		click.echo(">>> Goodbye...")

src/octoprint/plugins/softwareupdate/__init__.py

@@ -16,7 +16,7 @@ import logging
 import logging.handlers
 import hashlib
 
-from . import version_checks, updaters, exceptions, util
+from . import version_checks, updaters, exceptions, util, cli
 
 
 from octoprint.server.util.flask import restricted_access
@@ -351,7 +351,7 @@ class SoftwareUpdatePlugin(octoprint.plugin.BlueprintPlugin,
 	@restricted_access
 	def check_for_update(self):
 		if "check" in flask.request.values:
-			check_targets = map(str.strip, flask.request.values["check"].split(","))
+			check_targets = map(lambda x: x.strip(), flask.request.values["check"].split(","))
 		else:
 			check_targets = None
 
@@ -381,7 +381,7 @@ class SoftwareUpdatePlugin(octoprint.plugin.BlueprintPlugin,
 		json_data = flask.request.json
 
 		if "check" in json_data:
-			check_targets = map(str.strip, json_data["check"])
+			check_targets = map(lambda x: x.strip(), json_data["check"])
 		else:
 			check_targets = None
 
@@ -528,7 +528,8 @@ class SoftwareUpdatePlugin(octoprint.plugin.BlueprintPlugin,
 		updater_thread.daemon = False
 		updater_thread.start()
 
-		return to_be_updated, dict((key, check["displayName"] if "displayName" in check else key) for key, check in checks.items() if key in to_be_updated)
+		check_data = dict((key, self._populated_check(key, check)["displayName"]) for key, check in checks.items() if key in to_be_updated)
+		return to_be_updated, check_data
 
 	def _update_worker(self, checks, check_targets, force):
 
@@ -761,6 +762,7 @@ class SoftwareUpdatePlugin(octoprint.plugin.BlueprintPlugin,
 		else:
 			raise exceptions.UnknownUpdateType()
 
+
 __plugin_name__ = "Software Update"
 __plugin_author__ = "Gina Häußge"
 __plugin_url__ = "https://github.com/foosel/OctoPrint/wiki/Plugin:-Software-Update"
@@ -778,4 +780,9 @@ def __plugin_load__():
 		util=util
 	)
 
+	global __plugin_hooks__
+	__plugin_hooks__ = {
+		"octoprint.cli.commands": cli.commands
+	}
+
 

src/octoprint/plugins/softwareupdate/cli.py

@@ -0,0 +1,167 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+
+def commands(cli_group, pass_octoprint_ctx, *args, **kwargs):
+	import click
+	import sys
+	import requests.exceptions
+	import octoprint_client as client
+
+	@click.command("check")
+	@click.option("--force", is_flag=True, help="Ignore the cache for the update check")
+	@click.argument("targets", nargs=-1)
+	def check_command(force, targets):
+		"""Check for updates."""
+		params = dict(force=force)
+		if targets:
+			params["check"] = ",".join(targets)
+
+		client.init_client(cli_group.settings)
+		r = client.get("plugin/softwareupdate/check", params=params)
+		try:
+			r.raise_for_status()
+		except requests.exceptions.HTTPError as e:
+			click.echo("Could not get update information from server, got {}".format(e))
+			sys.exit(1)
+
+		data = r.json()
+		status = data["status"]
+		information = data["information"]
+
+		lines = []
+		octoprint_line = None
+		for key, info in information.items():
+			status_text = "Up to date"
+			if info["updateAvailable"]:
+				if info["updatePossible"]:
+					status_text = "Update available"
+				else:
+					status_text = "Update available (manual)"
+			line = "{}\n\tInstalled: {}\n\tAvailable: {}\n\t=> {}".format(info["displayName"],
+			                                                              info["information"]["local"]["name"],
+			                                                              info["information"]["remote"]["name"],
+			                                                              status_text)
+			if key == "octoprint":
+				octoprint_line = line
+			else:
+				lines.append(line)
+
+		lines.sort()
+		if octoprint_line:
+			lines = [octoprint_line] + lines
+
+		for line in lines:
+			click.echo(line)
+
+		click.echo()
+		if status == "current":
+			click.echo("Everything is up to date")
+		else:
+			click.echo("There are updates available!")
+
+
+	@click.command("update")
+	@click.option("--force", is_flag=True, help="Update even if already up to date")
+	@click.argument("targets", nargs=-1)
+	def update_command(force, targets):
+		"""Apply updates."""
+
+		data = dict(force=force)
+		if targets:
+			data["check"] = targets
+
+		client.init_client(cli_group.settings)
+
+		flags = dict(
+			waiting_for_restart=False,
+			seen_close=False
+		)
+
+		def on_message(ws, msg_type, msg):
+			if msg_type != "plugin" or msg["plugin"] != "softwareupdate":
+				return
+
+			plugin_message = msg["data"]
+			if not "type" in plugin_message:
+				return
+
+			plugin_message_type = plugin_message["type"]
+			plugin_message_data = plugin_message["data"]
+
+			if plugin_message_type == "updating":
+				click.echo("Updating {} to {}...".format(plugin_message_data["name"], plugin_message_data["target"]))
+
+			elif plugin_message_type == "update_failed":
+				click.echo("\t... failed: {}".format(plugin_message_data["reason"]))
+
+			elif plugin_message_type == "loglines" and "loglines" in plugin_message_data:
+				for entry in plugin_message_data["loglines"]:
+					prefix = ">>> " if entry["stream"] == "call" else ""
+					error = entry["stream"] == "stderr"
+					click.echo("\t{}{}".format(prefix, entry["line"].replace("\n", "\n\t")), err=error)
+
+			elif plugin_message_type == "success" or plugin_message_type == "restart_manually":
+				results = plugin_message_data["results"] if "results" in plugin_message_data else dict()
+				if results:
+					click.echo("The update finished successfully.")
+					if plugin_message_type == "restart_manually":
+						click.echo("Please restart the OctoPrint server.")
+				else:
+					click.echo("No update necessary")
+				ws.close()
+
+			elif plugin_message_type == "restarting":
+				flags["waiting_for_restart"] = True
+				click.echo("Restarting to apply changes...")
+
+			elif plugin_message_type == "failure":
+				click.echo("Error")
+				ws.close()
+
+		def on_open(ws):
+			if flags["waiting_for_restart"] and flags["seen_close"]:
+				click.echo(" Reconnected!")
+			else:
+				click.echo("Connected to server...")
+
+		def on_close(ws):
+			if flags["waiting_for_restart"] and flags["seen_close"]:
+				click.echo(".", nl=False)
+			else:
+				flags["seen_close"] = True
+				click.echo("Disconnected from server...")
+
+		socket = client.connect_socket(on_message=on_message,
+		                               on_open=on_open,
+		                               on_close=on_close)
+
+		r = client.post_json("plugin/softwareupdate/update", data=data)
+		try:
+			r.raise_for_status()
+		except requests.exceptions.HTTPError as e:
+			click.echo("Could not get update information from server, got {}".format(e))
+			sys.exit(1)
+
+		data = r.json()
+		to_be_updated = data["order"]
+		checks = data["checks"]
+		click.echo("Update in progress, updating:")
+		for name in to_be_updated:
+			click.echo("\t{}".format(name if not name in checks else checks[name]))
+
+		socket.wait()
+
+		if flags["waiting_for_restart"]:
+			if socket.reconnect(timeout=60):
+				click.echo("The update finished successfully.")
+			else:
+				click.echo("The update finished successfully but the server apparently didn't restart as expected.")
+				click.echo("Please restart the OctoPrint server.")
+
+	return [check_command, update_command]
+
+

src/octoprint_client/__init__.py

@@ -0,0 +1,384 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+
+import requests
+import time
+
+apikey = None
+baseurl = None
+
+
+class SocketTimeout(BaseException):
+	pass
+
+class SocketClient(object):
+	def __init__(self, url, use_ssl=False, daemon=True, **kwargs):
+		self._url = url
+		self._use_ssl = use_ssl
+		self._daemon = daemon
+		self._ws_kwargs = kwargs
+
+		self._seen_open = False
+		self._seen_close = False
+		self._waiting_for_reconnect = False
+
+		self._ws = None
+		self._thread = None
+
+		# hello world
+
+	def _prepare(self):
+		"""Prepares socket and thread for a new connection."""
+
+		# close the socket if it's currently open
+		if self._ws is not None:
+			try:
+				self._ws.close()
+			except:
+				# we can't handle that in any meaningful way right now
+				pass
+
+		# prepare a bunch of callback methods
+		import functools
+		callbacks = dict()
+		for callback in ("on_open", "on_message", "on_error", "on_close"):
+			# now, normally we could just use functools.partial for something like
+			# this, but websocket does a type check against a python function type
+			# which the use of partial makes fail, so we have to use a lambda
+			# wrapper here, and since we need to pick the current value of
+			# callback from the current scope we need a factory method for that...
+			def factory(cb):
+				return lambda *fargs, **fkwargs: functools.partial(self._on_callback, cb)(*fargs, **fkwargs)
+			callbacks[callback] = factory(callback)
+
+		# initialize socket instance with url and callbacks
+		import websocket
+		kwargs = dict(self._ws_kwargs)
+		kwargs.update(callbacks)
+		self._ws = websocket.WebSocketApp(self._url, **kwargs)
+
+		# initialize thread
+		import threading
+		self._thread = threading.Thread(target=self._on_thread_run)
+		self._thread.daemon = self._daemon
+
+	def _on_thread_run(self):
+		"""Has the socket run forever (aka until closed...)."""
+		self._ws.run_forever()
+
+	def _on_callback(self, cb, *args, **kwargs):
+		"""
+		Callback for socket events.
+
+		Will call any callback method defined on ``self`` that matches ``cb``
+		prefixed with an ``_ws_``, then will call any callback method provided in
+		the socket keyword arguments (``self._ws_kwargs``) that matches ``cb``.
+
+		Calling args and kwargs will be the ones passed to ``_on_callback``.
+
+		Arguments:
+		    cb (str): the callback type
+		"""
+		internal = "_ws_" + cb
+		if hasattr(self, internal):
+			cb_func = getattr(self, internal)
+			if callable(cb_func):
+				cb_func(*args, **kwargs)
+
+		cb_func = self._ws_kwargs.get(cb, None)
+		if callable(cb_func):
+			cb_func(*args, **kwargs)
+
+	def _ws_on_open(self, ws):
+		"""
+		Callback for socket on_open event.
+
+		Used only to track active reconnection attempts.
+		"""
+
+		if not self._waiting_for_reconnect:
+			return
+		if ws != self._ws:
+			return
+		self._seen_open = True
+
+	def _ws_on_close(self, ws):
+		"""
+		Callback for socket on_close event.
+
+		Used only to track active reconnection attempts.
+		"""
+
+		if not self._waiting_for_reconnect:
+			return
+		if ws != self._ws:
+			return
+		self._seen_close = True
+
+	def connect(self):
+		"""Connects the socket."""
+		self._prepare()
+		self._thread.start()
+
+	def wait(self, timeout=None):
+		"""Waits for the closing of the socket or the timeout."""
+		start = None
+
+		def test_condition():
+			if timeout and start and start + timeout > time.time():
+				raise SocketTimeout()
+
+		start = time.time()
+		while self._thread.is_alive():
+			test_condition()
+			self._thread.join(timeout=1.0)
+
+	@property
+	def is_connected(self):
+		"""Whether the web socket is connected or not."""
+		return self._thread and self._ws and self._thread.is_alive()
+
+	def disconnect(self):
+		"""Disconnect the web socket."""
+		if self._ws:
+			self._ws.close()
+
+	def reconnect(self, timeout=None, disconnect=True):
+		"""
+		Tries to reconnect to the web socket.
+
+		If timeout is set will try to reconnect over the specified timeout in seconds
+		and return False if the connection could not be re-established.
+
+		If no timeout is provided, the method will block until the connection could
+		be re-established.
+
+		If disconnect is set to ``True`` will disconnect the socket explictly
+		first if it is currently connected.
+
+		Arguments:
+		    timeout (number): timeout in seconds to wait for the reconnect to happen.
+		    disconnect (bool): Whether to disconnect explicitly from the socket if
+		       a connection is currently established (True, default) or not (False).
+
+		Returns:
+		    bool - True if the reconnect was successful, False otherwise.
+		"""
+
+		self._seen_close = False
+		self._seen_open = False
+		self._waiting_for_reconnect = True
+
+		if not self.is_connected:
+			# not connected, so we are already closed
+			self._seen_close = True
+		elif disconnect:
+			# connected and disconnect is True, so we disconnect
+			self.disconnect()
+
+		start = None
+		if timeout:
+			timeout_condition = lambda: start is not None and time.time() > start + timeout
+		else:
+			timeout_condition = lambda: False
+
+		start = time.time()
+		while not timeout_condition():
+			if self._seen_close and self._seen_open:
+				# we saw a connection close and open, so a reconnect, success!
+				return True
+			else:
+				# try to connect
+				self.connect()
+
+			# sleep a bit
+			time.sleep(1.0)
+
+		# if we land here the timeout condition became True without us seeing
+		# a reconnect, that's a failure
+		return False
+
+
+def build_base_url(https=False, httpuser=None, httppass=None, host=None, port=None, prefix=None):
+	protocol = "https" if https else "http"
+	httpauth = "{}:{}@".format(httpuser, httppass) if httpuser and httppass else ""
+	host = host if host else "127.0.0.1"
+	port = ":{}".format(port) if port else ":5000"
+	prefix = prefix if prefix else ""
+
+	return "{}://{}{}{}{}".format(protocol, httpauth, host, port, prefix)
+
+
+def init_client(settings, https=False, httpuser=None, httppass=None, host=None, port=None, prefix=None):
+	"""
+	Initializes the API client with the provided settings.
+
+	Basically a convenience method to set ``apikey`` and ``baseurl`` from settings
+	and/or command line arguments.
+
+	Arguments:
+	    settings (octoprint.settings.Settings): A :class:`~octoprint.settings.Settings` instance to use
+	        for client configuration
+	    https (bool): Whether to connect via HTTPS (True) or not (False, default)
+	    httpuser (str or None): HTTP Basic Auth username to use. No Basic Auth will be
+	        used if unset.
+	    httppass (str or None): HTTP Basic Auth password to use. No Basic Auth will be
+	        used if unset.
+	    host (str or None): Host to connect to, overrides data from settings if set.
+	    port (int or None): Port to connect to, overrides data from settings if set.
+	    prefix (str or None): Path prefix, overrides data from settings if set.
+	"""
+	settings_host = settings.get(["server", "host"])
+	settings_port = settings.getInt(["server", "port"])
+	settings_apikey = settings.get(["api", "key"])
+
+	global apikey, baseurl
+	apikey = settings_apikey
+	baseurl = build_base_url(https=https,
+	                         httpuser=httpuser,
+	                         httppass=httppass,
+	                         host=host or settings_host if settings_host != "0.0.0.0" else "127.0.0.1",
+	                         port=port or settings_port,
+	                         prefix=prefix)
+
+def prepare_request(method=None, path=None, params=None):
+	url = None
+	if baseurl:
+		while path.startswith("/"):
+			path = path[1:]
+		url = baseurl + "/" + path
+	return requests.Request(method=method, url=url, params=params, headers={"X-Api-Key": apikey}).prepare()
+
+def request(method, path, data=None, files=None, encoding=None, params=None):
+	s = requests.Session()
+	request = prepare_request(method, path, params=params)
+	if data or files:
+		if encoding == "json":
+			request.prepare_body(None, None, json=data)
+		else:
+			request.prepare_body(data, files=files)
+	response = s.send(request)
+	return response
+
+def get(path, params=None):
+	return request("GET", path, params=params)
+
+def post(path, data, encoding=None, params=None):
+	return request("POST", path, data=data, encoding=encoding, params=params)
+
+def post_json(path, data, params=None):
+	return post(path, data, encoding="json", params=params)
+
+def post_command(path, command, additional=None):
+	data = dict(command=command)
+	if additional:
+		data.update(additional)
+	return post_json(path, data, params=data)
+
+def upload(path, file_path, additional=None, file_name=None, content_type=None, params=None):
+	import os
+
+	if not os.path.isfile(file_path):
+		raise ValueError("{} cannot be uploaded since it is not a file".format(file_path))
+
+	if file_name is None:
+		file_name = os.path.basename(file_path)
+
+	with open(file_path, "rb") as fp:
+		if content_type:
+			files = dict(file=(file_name, fp, content_type))
+		else:
+			files = dict(file=(file_name, fp))
+
+		response = request("POST", path, data=additional, files=files, params=params)
+
+	return response
+
+def delete(path, params=None):
+	return request("DELETE", path, params=params)
+
+def patch(path, data, encoding=None, params=None):
+	return request("PATCH", path, data=data, encoding=encoding, params=params)
+
+def put(path, data, encoding=None, params=None):
+	return request("PUT", path, data=data, encoding=encoding, params=params)
+
+def connect_socket(**kwargs):
+	import uuid
+	import random
+	import json
+
+	# creates websocket URL for SockJS according to
+	# - http://sockjs.github.io/sockjs-protocol/sockjs-protocol-0.3.3.html#section-37
+	# - http://sockjs.github.io/sockjs-protocol/sockjs-protocol-0.3.3.html#section-50
+	url = "ws://{}/sockjs/{:0>3d}/{}/websocket".format(
+		baseurl[baseurl.find("//") + 2:], # host + port + prefix, but no protocol
+		random.randrange(0, stop=999),    # server_id
+		uuid.uuid4()                      # session_id
+	)
+	use_ssl = baseurl.startswith("https:")
+
+	on_open_cb = kwargs.get("on_open", None)
+	on_heartbeat_cb = kwargs.get("on_heartbeat", None)
+	on_message_cb = kwargs.get("on_message", None)
+	on_close_cb = kwargs.get("on_close", None)
+	on_error_cb = kwargs.get("on_error", None)
+	daemon = kwargs.get("daemon", True)
+
+	def on_message(ws, message):
+		message_type = message[0]
+
+		if message_type == "h":
+			# "heartbeat" message
+			if callable(on_heartbeat_cb):
+				on_heartbeat_cb(ws)
+				return
+		elif message_type == "o":
+			# "open" message
+			return
+		elif message_type == "c":
+			# "close" message
+			return
+
+		if not callable(on_message_cb):
+			return
+
+		message_body = message[1:]
+		if not message_body:
+			return
+
+		data = json.loads(message_body)
+
+		if message_type == "m":
+			data = [data,]
+
+		for d in data:
+			for internal_type, internal_message in d.items():
+				on_message_cb(ws, internal_type, internal_message)
+
+	def on_open(ws):
+		if callable(on_open_cb):
+			on_open_cb(ws)
+
+	def on_close(ws):
+		if callable(on_close_cb):
+			on_close_cb(ws)
+
+	def on_error(ws, error):
+		if callable(on_error_cb):
+			on_error_cb(ws, error)
+
+	socket = SocketClient(url,
+	                      use_ssl=use_ssl,
+	                      daemon=daemon,
+	                      on_open=on_open,
+	                      on_message=on_message,
+	                      on_close=on_close,
+	                      on_error=on_error)
+	socket.connect()
+
+	return socket