PrivateFunction: _connect_cb_wrapper
_Private_ function that handles the first connection messages.

On receiving an opening stream tag this callback replaces itself with the real
message handler. On receiving a stream error the connection is terminated.
960668
function(message) {
        if (message.data.indexOf("<open ") === 0 || message.data.indexOf("<?xml") === 0) {
            // Strip the XML Declaration, if there is one
            var data = message.data.replace(/^(<\?.*?\?>\s*)*/, "");
            if (data === '') return;

            var streamStart = new DOMParser().parseFromString(data, "text/xml").documentElement;
            this._conn.xmlInput(streamStart);
            this._conn.rawInput(message.data);

            //_handleStreamSteart will check for XML errors and disconnect on error
            if (this._handleStreamStart(streamStart)) {
                //_connect_cb will check for stream:error and disconnect on error
                this._connect_cb(streamStart);
            }
        } else if (message.data.indexOf("<close ") === 0) { //'<close xmlns="urn:ietf:params:xml:ns:xmpp-framing />') {
            this._conn.rawInput(message.data);
            this._conn.xmlInput(message);
            var see_uri = message.getAttribute("see-other-uri");
            if (see_uri) {
                this._conn._changeConnectStatus(Strophe.Status.REDIRECT, "Received see-other-uri, resetting connection");
                this._conn.reset();
                this._conn.service = see_uri;
                this._connect();
            } else {
                this._conn._changeConnectStatus(Strophe.Status.CONNFAIL, "Received closing stream");
                this._conn._doDisconnect();
            }
        } else {
            var string = this._streamWrap(message.data);
            var elem = new DOMParser().parseFromString(string, "text/xml").documentElement;
            this.socket.onmessage = this._onMessage.bind(this);
            this._conn._connect_cb(elem, null, message.data);
        }
    }
1218324
def on_receive(self, message=None, wire=None, event_origin=None):

        

        self.trigger("before_call", message)

        fn_name = message.data
        pmsg = self.prepare_message

        try:
            for handler in self.handlers:
                handler.incoming(message, self)

            fn = self.get_function(fn_name, message.path)
        except Exception as inst:
            wire.respond(message, ErrorMessage(str(inst)))
            return

        if callable(fn) and getattr(fn, "exposed", False):
            try:
                r = fn(*message.args, **message.kwargs)
                if isinstance(r,Message):
                    wire.respond(message, pmsg(r))
                else:
                    wire.respond(message, pmsg(Message(r)))
            except Exception as inst:
                if self.debug:
                    wire.respond(message, pmsg(ErrorMessage(str(traceback.format_exc()))))
                else:
                    wire.respond(message, pmsg(ErrorMessage(str(inst))))
        else:
            wire.respond(
                message,
                pmsg(
                    ErrorMessage("action '%s' not exposed on API (%s)" % (fn_name, self.__class__.__name__)))
                )

        self.trigger("after_call", message)
146808
async function close(callback) {
            await closeStream(client.stream);
            await closeStream(client.rpc);
            if (callback) {
                return callback();
            }
        }
201102
function(cb) {
    // return early if already connected
    if (_socket) {
      return cb(null, {host: _options.host, port: _options.port});
    }

    _socket = net.createConnection(_options.port, _options.host);

    _socket.on('connect', function() {
      if (cb) {
        cb(null, {host: _options.host, port: _options.port});
      }
    });

    _socket.on('data', function(buf) {
      var res;
      var rblock;
      var decoder;

      if (!_rcvBuf) {
        _rcvBuf = kb();
      }
      _rcvBuf.append(buf);
      res = response.decodeHead(_rcvBuf.get());
      // console.log(res)

      // buffer can have more than one msg...
      // don't want to drop msgs
      while (_rcvBuf.length() > 0 && _rcvBuf.length() >= res.size + 4) {
        decoder = _cbt.remove(res.correlation);
        rblock = decoder.decode(_rcvBuf.get(), res);

        _rcvBuf.slice(res.size + 4);

        if (_rcvBuf.length() > 0) {
          res = response.decodeHead(_rcvBuf.get());
        }

        decoder.callback(rblock.err, rblock.result);
      }
    });

    _socket.on('end', function() {});

    _socket.on('timeout', function(){
      console.log('socket timeout');
    });

    _socket.on('drain', function(){});

    _socket.on('error', function(err){
      console.log('SOCKET ERROR: ' + err);
    });

    _socket.on('close', function(){});
  }
139088
public void connect(
		final String endpoint,
		final Callback<? super HttpSocket> onConnect,
		final Callback<? super Exception> onError
	) {
		executors.getUnbounded().submit(
			new Runnable() {
				@Override
				public void run() {
					try {
						// Build request bytes
						AoByteArrayOutputStream bout = new AoByteArrayOutputStream();
						try {
							try (DataOutputStream out = new DataOutputStream(bout)) {
								out.writeBytes("action=connect");
							}
						} finally {
							bout.close();
						}
						long connectTime = System.currentTimeMillis();
						URL endpointURL = new URL(endpoint);
						HttpURLConnection conn = (HttpURLConnection)endpointURL.openConnection();
						conn.setAllowUserInteraction(false);
						conn.setConnectTimeout(CONNECT_TIMEOUT);
						conn.setDoOutput(true);
						conn.setFixedLengthStreamingMode(bout.size());
						conn.setInstanceFollowRedirects(false);
						conn.setReadTimeout(CONNECT_TIMEOUT);
						conn.setRequestMethod("POST");
						conn.setUseCaches(false);
						// Write request
						OutputStream out = conn.getOutputStream();
						try {
							out.write(bout.getInternalByteArray(), 0, bout.size());
							out.flush();
						} finally {
							out.close();
						}
						// Get response
						int responseCode = conn.getResponseCode();
						if(responseCode != 200) throw new IOException("Unexpect response code: " + responseCode);
						if(DEBUG) System.out.println("DEBUG: HttpSocketClient: connect: got connection");
						DocumentBuilder builder = builderFactory.newDocumentBuilder();
						Element document = builder.parse(conn.getInputStream()).getDocumentElement();
						if(!"connection".equals(document.getNodeName())) throw new IOException("Unexpected root node name: " + document.getNodeName());
						Identifier id = Identifier.valueOf(document.getAttribute("id"));
						if(DEBUG) System.out.println("DEBUG: HttpSocketClient: connect: got id=" + id);
						HttpSocket httpSocket = new HttpSocket(
							HttpSocketClient.this,
							id,
							connectTime,
							endpointURL
						);
						if(DEBUG) System.out.println("DEBUG: HttpSocketClient: connect: adding socket");
						addSocket(httpSocket);
						if(onConnect!=null) {
							if(DEBUG) System.out.println("DEBUG: HttpSocketClient: connect: calling onConnect");
							onConnect.call(httpSocket);
						}
					} catch(Exception exc) {
						if(onError!=null) {
							if(DEBUG) System.out.println("DEBUG: HttpSocketClient: connect: calling onError");
							onError.call(exc);
						}
					}
				}
			}
		);
	}
220513
function wrapConnection(conn) {
    // Already wrapped this function?
    if (conn._coWrapped) {
        return conn;
    }
        
    // Set flag to avoid re-wrapping
    conn._coWrapped = true;

    // Functions to thunkify
    var queryMethods = [
        'query',
        'execute'
    ];
    
    // Traverse query methods list
    queryMethods.forEach(function(name) {
        // Thunkify the method
        conn[name] = wrapQueryMethod(conn[name], conn);
    });
    
    // Return co-friendly connection
    return conn;
}
663336
function (options, callback) {
            /// <summary>Starts the connection</summary>
            /// <param name="options" type="Object">Options map</param>
            /// <param name="callback" type="Function">A callback function to execute when the connection has started</param>
            var connection = this,
                config = {
                    pingInterval: 300000,
                    waitForPageLoad: true,
                    transport: "auto",
                    jsonp: false
                },
                initialize,
                deferred = connection._deferral || $.Deferred(), // Check to see if there is a pre-existing deferral that's being built on, if so we want to keep using it
                parser = window.document.createElement("a");

            connection.lastError = null;

            // Persist the deferral so that if start is called multiple times the same deferral is used.
            connection._deferral = deferred;

            if (!connection.json) {
                // no JSON!
                throw new Error("SignalR: No JSON parser found. Please ensure json2.js is referenced before the SignalR.js file if you need to support clients without native JSON parsing support, e.g. IE<8.");
            }

            if ($.type(options) === "function") {
                // Support calling with single callback parameter
                callback = options;
            } else if ($.type(options) === "object") {
                $.extend(config, options);
                if ($.type(config.callback) === "function") {
                    callback = config.callback;
                }
            }

            config.transport = validateTransport(config.transport, connection);

            // If the transport is invalid throw an error and abort start
            if (!config.transport) {
                throw new Error("SignalR: Invalid transport(s) specified, aborting start.");
            }

            connection._.config = config;

            // Check to see if start is being called prior to page load
            // If waitForPageLoad is true we then want to re-direct function call to the window load event
            if (!_pageLoaded && config.waitForPageLoad === true) {
                connection._.deferredStartHandler = function () {
                    connection.start(options, callback);
                };
                _pageWindow.bind("load", connection._.deferredStartHandler);

                return deferred.promise();
            }

            // If we're already connecting just return the same deferral as the original connection start
            if (connection.state === signalR.connectionState.connecting) {
                return deferred.promise();
            } else if (changeState(connection,
                            signalR.connectionState.disconnected,
                            signalR.connectionState.connecting) === false) {
                // We're not connecting so try and transition into connecting.
                // If we fail to transition then we're either in connected or reconnecting.

                deferred.resolve(connection);
                return deferred.promise();
            }

            configureStopReconnectingTimeout(connection);

            // Resolve the full url
            parser.href = connection.url;
            if (!parser.protocol || parser.protocol === ":") {
                connection.protocol = window.document.location.protocol;
                connection.host = parser.host || window.document.location.host;
            } else {
                connection.protocol = parser.protocol;
                connection.host = parser.host;
            }

            connection.baseUrl = connection.protocol + "//" + connection.host;

            // Set the websocket protocol
            connection.wsProtocol = connection.protocol === "https:" ? "wss://" : "ws://";

            // If jsonp with no/auto transport is specified, then set the transport to long polling
            // since that is the only transport for which jsonp really makes sense.
            // Some developers might actually choose to specify jsonp for same origin requests
            // as demonstrated by Issue #623.
            if (config.transport === "auto" && config.jsonp === true) {
                config.transport = "longPolling";
            }

            // If the url is protocol relative, prepend the current windows protocol to the url. 
            if (connection.url.indexOf("//") === 0) {
                connection.url = window.location.protocol + connection.url;
                connection.log("Protocol relative URL detected, normalizing it to '" + connection.url + "'.");
            }

            if (this.isCrossDomain(connection.url)) {
                connection.log("Auto detected cross domain url.");

                if (config.transport === "auto") {
                    // TODO: Support XDM with foreverFrame
                    config.transport = ["webSockets", "serverSentEvents", "longPolling"];
                }

                if (typeof (config.withCredentials) === "undefined") {
                    config.withCredentials = true;
                }

                // Determine if jsonp is the only choice for negotiation, ajaxSend and ajaxAbort.
                // i.e. if the browser doesn't supports CORS
                // If it is, ignore any preference to the contrary, and switch to jsonp.
                if (!config.jsonp) {
                    config.jsonp = !$.support.cors;

                    if (config.jsonp) {
                        connection.log("Using jsonp because this browser doesn't support CORS.");
                    }
                }

                connection.contentType = signalR._.defaultContentType;
            }

            connection.withCredentials = config.withCredentials;

            connection.ajaxDataType = config.jsonp ? "jsonp" : "text";

            $(connection).bind(events.onStart, function (e, data) {
                if ($.type(callback) === "function") {
                    callback.call(connection);
                }
                deferred.resolve(connection);
            });

            connection._.initHandler = signalR.transports._logic.initHandler(connection);

            initialize = function (transports, index) {
                var noTransportError = signalR._.error(resources.noTransportOnInit);

                index = index || 0;
                if (index >= transports.length) {
                    if (index === 0) {
                        connection.log("No transports supported by the server were selected.");
                    } else if (index === 1) {
                        connection.log("No fallback transports were selected.");
                    } else {
                        connection.log("Fallback transports exhausted.");
                    }

                    // No transport initialized successfully
                    $(connection).triggerHandler(events.onError, [noTransportError]);
                    deferred.reject(noTransportError);
                    // Stop the connection if it has connected and move it into the disconnected state
                    connection.stop();
                    return;
                }

                // The connection was aborted
                if (connection.state === signalR.connectionState.disconnected) {
                    return;
                }

                var transportName = transports[index],
                    transport = signalR.transports[transportName],
                    onFallback = function () {
                        initialize(transports, index + 1);
                    };

                connection.transport = transport;

                try {
                    connection._.initHandler.start(transport, function () { // success
                        // Firefox 11+ doesn't allow sync XHR withCredentials: https://developer.mozilla.org/en-US/docs/Web/API/XMLHttpRequest#withCredentials
                        var isFirefox11OrGreater = signalR._.firefoxMajorVersion(window.navigator.userAgent) >= 11,
                            asyncAbort = !!connection.withCredentials && isFirefox11OrGreater;

                        connection.log("The start request succeeded. Transitioning to the connected state.");

                        if (supportsKeepAlive(connection)) {
                            signalR.transports._logic.monitorKeepAlive(connection);
                        }

                        signalR.transports._logic.startHeartbeat(connection);

                        // Used to ensure low activity clients maintain their authentication.
                        // Must be configured once a transport has been decided to perform valid ping requests.
                        signalR._.configurePingInterval(connection);

                        if (!changeState(connection,
                                            signalR.connectionState.connecting,
                                            signalR.connectionState.connected)) {
                            connection.log("WARNING! The connection was not in the connecting state.");
                        }

                        // Drain any incoming buffered messages (messages that came in prior to connect)
                        connection._.connectingMessageBuffer.drain();

                        $(connection).triggerHandler(events.onStart);

                        // wire the stop handler for when the user leaves the page
                        _pageWindow.bind("unload", function () {
                            connection.log("Window unloading, stopping the connection.");

                            connection.stop(asyncAbort);
                        });

                        if (isFirefox11OrGreater) {
                            // Firefox does not fire cross-domain XHRs in the normal unload handler on tab close.
                            // #2400
                            _pageWindow.bind("beforeunload", function () {
                                // If connection.stop() runs runs in beforeunload and fails, it will also fail
                                // in unload unless connection.stop() runs after a timeout.
                                window.setTimeout(function () {
                                    connection.stop(asyncAbort);
                                }, 0);
                            });
                        }
                    }, onFallback);
                }
                catch (error) {
                    connection.log(transport.name + " transport threw '" + error.message + "' when attempting to start.");
                    onFallback();
                }
            };

            var url = connection.url + "/negotiate",
                onFailed = function (error, connection) {
                    var err = signalR._.error(resources.errorOnNegotiate, error, connection._.negotiateRequest);

                    $(connection).triggerHandler(events.onError, err);
                    deferred.reject(err);
                    // Stop the connection if negotiate failed
                    connection.stop();
                };

            $(connection).triggerHandler(events.onStarting);

            url = signalR.transports._logic.prepareQueryString(connection, url);

            connection.log("Negotiating with '" + url + "'.");

            // Save the ajax negotiate request object so we can abort it if stop is called while the request is in flight.
            connection._.negotiateRequest = signalR.transports._logic.ajax(connection, {
                url: url,
                error: function (error, statusText) {
                    // We don't want to cause any errors if we're aborting our own negotiate request.
                    if (statusText !== _negotiateAbortText) {
                        onFailed(error, connection);
                    } else {
                        // This rejection will noop if the deferred has already been resolved or rejected.
                        deferred.reject(signalR._.error(resources.stoppedWhileNegotiating, null /* error */, connection._.negotiateRequest));
                    }
                },
                success: function (result) {
                    var res,
                        keepAliveData,
                        protocolError,
                        transports = [],
                        supportedTransports = [];

                    try {
                        res = connection._parseResponse(result);
                    } catch (error) {
                        onFailed(signalR._.error(resources.errorParsingNegotiateResponse, error), connection);
                        return;
                    }

                    keepAliveData = connection._.keepAliveData;
                    connection.appRelativeUrl = res.Url;
                    connection.id = res.ConnectionId;
                    connection.token = res.ConnectionToken;
                    connection.webSocketServerUrl = res.WebSocketServerUrl;

                    // The long poll timeout is the ConnectionTimeout plus 10 seconds
                    connection._.pollTimeout = res.ConnectionTimeout * 1000 + 10000; // in ms

                    // Once the server has labeled the PersistentConnection as Disconnected, we should stop attempting to reconnect
                    // after res.DisconnectTimeout seconds.
                    connection.disconnectTimeout = res.DisconnectTimeout * 1000; // in ms

                    // Add the TransportConnectTimeout from the response to the transportConnectTimeout from the client to calculate the total timeout
                    connection._.totalTransportConnectTimeout = connection.transportConnectTimeout + res.TransportConnectTimeout * 1000;

                    // If we have a keep alive
                    if (res.KeepAliveTimeout) {
                        // Register the keep alive data as activated
                        keepAliveData.activated = true;

                        // Timeout to designate when to force the connection into reconnecting converted to milliseconds
                        keepAliveData.timeout = res.KeepAliveTimeout * 1000;

                        // Timeout to designate when to warn the developer that the connection may be dead or is not responding.
                        keepAliveData.timeoutWarning = keepAliveData.timeout * connection.keepAliveWarnAt;

                        // Instantiate the frequency in which we check the keep alive.  It must be short in order to not miss/pick up any changes
                        connection._.beatInterval = (keepAliveData.timeout - keepAliveData.timeoutWarning) / 3;
                    } else {
                        keepAliveData.activated = false;
                    }

                    connection.reconnectWindow = connection.disconnectTimeout + (keepAliveData.timeout || 0);

                    if (!res.ProtocolVersion || res.ProtocolVersion !== connection.clientProtocol) {
                        protocolError = signalR._.error(signalR._.format(resources.protocolIncompatible, connection.clientProtocol, res.ProtocolVersion));
                        $(connection).triggerHandler(events.onError, [protocolError]);
                        deferred.reject(protocolError);

                        return;
                    }

                    $.each(signalR.transports, function (key) {
                        if ((key.indexOf("_") === 0) || (key === "webSockets" && !res.TryWebSockets)) {
                            return true;
                        }
                        supportedTransports.push(key);
                    });

                    if ($.isArray(config.transport)) {
                        $.each(config.transport, function (_, transport) {
                            if ($.inArray(transport, supportedTransports) >= 0) {
                                transports.push(transport);
                            }
                        });
                    } else if (config.transport === "auto") {
                        transports = supportedTransports;
                    } else if ($.inArray(config.transport, supportedTransports) >= 0) {
                        transports.push(config.transport);
                    }

                    initialize(transports);
                }
            });

            return deferred.promise();
        }
242782
function(callback){
        callback = callback || noop;

        if(!this._client){
            return callback();
        }
        this._client.shutdown(function(err){
            if(!this._define_connection){
                return callback(err);
            }
            this._define_connection.shutdown(function(derr){
                callback(err || derr);
            });
        }.bind(this));
    }
5930
function (url, callback) {
		if (_websocket) {
			_websocket.close();
			_websocket = null;
		}
		var ws = _websocket = new WebSocket(url || DEFAULT_URL);
		ws.onopen = function (event) {
			trigger("open", event);
		};
		ws.onerror = function (event) {
			trigger("error", event);
		};
		ws.onclose = function (event) {
			trigger("close", event);
		};
		var handler = function (event) {
			reset_idle_ping_timer();
			var msg = JSON.parse(event.data);
			console.log("Coinfloor >>", msg);
			if ("tag" in msg) {
				var handler = _result_handlers[msg.tag];
				delete _result_handlers[msg.tag];
				if (handler) {
					handler(msg);
				}
				else if (handler === undefined) {
					console.error("Coinfloor:", msg);
				}
			}
			else if ("notice" in msg) {
				trigger(msg.notice, msg);
			}
		};
		ws.onmessage = function (event) {
			var msg = JSON.parse(event.data);
			_server_nonce = atob(msg.nonce);
			this.onmessage = handler;
			handler(event);
			if (callback) {
				callback(msg);
			}
		};
	}
61415
function() {
    // choose from the sub-protocols
    if (typeof self.options.handleProtocols == 'function') {
      var protList = (protocols || '').split(/, */);
      var callbackCalled = false;
      self.options.handleProtocols(protList, function(result, protocol) {
        callbackCalled = true;
        if (!result) abortConnection(socket, 401, 'Unauthorized');
        else completeHybiUpgrade2(protocol);
      });
      if (!callbackCalled) {
        // the handleProtocols handler never called our callback
        abortConnection(socket, 501, 'Could not process protocols');
      }
      return;
    } else {
      if (typeof protocols !== 'undefined') {
        completeHybiUpgrade2(protocols.split(/, */)[0]);
      }
      else {
        completeHybiUpgrade2();
      }
    }
  }
813303
function(_connection) {
    return function() {
      // Destroyed state return
      if (self.state === DESTROYED || self.state === DESTROYING) {
        // Remove the connection from the list
        removeConnection(self, _connection);
        return _connection.destroy();
      }

      // Destroy all event emitters
      handlers.forEach(function(e) {
        _connection.removeAllListeners(e);
      });

      // Add the final handlers
      _connection.once('close', connectionFailureHandler(self, 'close'));
      _connection.once('error', connectionFailureHandler(self, 'error'));
      _connection.once('timeout', connectionFailureHandler(self, 'timeout'));
      _connection.once('parseError', connectionFailureHandler(self, 'parseError'));

      // Signal
      reauthenticate(self, _connection, function(err) {
        if (self.state === DESTROYED || self.state === DESTROYING) {
          return _connection.destroy();
        }
        // Remove the connection from the connectingConnections list
        removeConnection(self, _connection);

        // Handle error
        if (err) {
          return _connection.destroy();
        }

        // If we are c at the moment
        // Do not automatially put in available connections
        // As we need to apply the credentials first
        if (self.authenticating) {
          self.nonAuthenticatedConnections.push(_connection);
        } else {
          // Push to available
          self.availableConnections.push(_connection);
          // Execute any work waiting
          _execute(self)();
        }
      });
    };
  }
14577
function ({ live }, cb) {
      var stream = PullPushAbort()

      var result = {}
      var sync = null

      pull(
        ssb.query.read({
          query: [{ $filter: {
            value: {
              author: ssb.id,
              content: {
                type: 'contact'
              }
            }
          } }],
          private: true,
          live
        }),
        stream.aborter,
        pull.drain((msg) => {
          if (msg.sync) {
            sync = true
            if (live) {
              stream.push(result)
            }
            return
          }

          var isPrivate = msg.value && msg.value.meta && msg.value.meta.private
          var content = msg.value.content

          // if a non-private state has been set since last private, revert to null
          // patchwork will always try to set a new ignore status after the public if it differs
          // this is just to handle the case where the private state was set by another client
          // (which will override ignore due to the way ssb-friends handles private blocks)

          var value = isPrivate ? !!content.blocking : null

          if (sync) {
            stream.push({ [content.contact]: value })
          } else {
            result[content.contact] = value
          }
        }, (err) => {
          if (err) return stream.end(err)
          if (stream.ended || sync) return

          if (!live) {
            stream.push(result)
          }

          stream.end()
        })
      )

      return stream
    }
262464
function Connection(endpoint, options, connectionListener) {
  if (!(this instanceof Connection)) return new Connection(endpoint, options, connectionListener);

  stream.Duplex.call(this);

  if (endpoint && !(endpoint instanceof Endpoint)) {
    endpoint = new Endpoint(endpoint);
  }
  if (typeof options === 'function') {
    connectionListener = options;
    options = undefined;
  }
  this.endpoint = endpoint;
  this.options = options || {};
  if (typeof connectionListener === 'function') {
    this.on('connect', connectionListener);
  }
  this.status = {sent: 0, received: 0};

  this.on('_read_data', function (data) {
    this.push(data);
    this.status.received += data.length;
  });

  if (endpoint) {
    //connect
    connect.call(this);
  }
}
1023500
function (obj) {
			if (typeof connHandler === 'function')
				connHandler('disconnect');
			
			SR.Warn('disconnected... obj: ');
			SR.Warn(obj);
			
			// attempt to re-connect
			if (typeof conn_options === 'object') {
				removeEntry(currentEntry);
				currentEntry = undefined;
				SR.Warn('re-connecting websocket with options');
				SR.Warn(conn_options);
				SR.setSocketServer(conn_options);
			}
		}
69399
AmqpClient closeOkConnection(Continuation callback, ErrorHandler error) {
        Object[] args = {};
        AmqpBuffer bodyArg = null;

        String methodName = "closeOkConnection";
        String methodId = "10" + "50";
        AmqpMethod amqpMethod = MethodLookup.LookupMethod(methodId);
        Object[] arguments = {this, amqpMethod, this.id, args, bodyArg};
        asyncClient.enqueueAction(methodName, "write", arguments, callback, error);
        return this;
    }
1090370
function(client, callback) {
    logger.entry('Client.stop.performDisconnect', client.id);

    // If this client is still starting, then wait until the state changes
    // before we attempt to stop.
    if (client.state === STATE_STARTING) {
      var stillConnecting = function(client) {
        logger.entry('stillConnecting', client.id);

        if (client.state === STATE_STARTING) {
          setImmediate(function() {
            stillConnecting(client);
          });
        } else {
          process.nextTick(function() {
            performDisconnect(client, callback);
          });
        }

        logger.exit('stillConnecting', client.id, null);
      };

      setImmediate(function() {
        stillConnecting(client);
      });
      logger.exit('Client.stop.performDisconnect', client.id, null);
      return;
    }

    client._setState(STATE_STOPPING);

    if (client._serviceFunctionTimeout) {
      // if we previously set a timeout to retry the serviceFunction, clear it.
      clearTimeout(client._serviceFunctionTimeout);
    }
    client._serviceFunctionTimeout = null;

    // Only disconnect when all outstanding send operations are complete
    if (client._outstandingSends.length === 0) {
      stopMessenger(client, function(client, callback) {
        logger.entry('Client.stop.performDisconnect.stopProcessing',
            client.id);
        // clear queuedSends as we are disconnecting
        var messages = [];
        while (client._queuedSends.length > 0) {
          messages.push(client._queuedSends.shift());
        }
        // call the msg callbacks in error as we have disconnected
        process.nextTick(function() {
          logger.entry('Client.stop.performDisconnect.' +
              'stopProcessing.queuedSendCallback', client.id);
          while (messages.length > 0) {
            var msg = messages.shift();
            if (msg.callback) {
              msg.callback(new StoppedError('send aborted due to client stop'));
            }
          }
          logger.exit('Client.stop.performDisconnect.' +
              'stopProcessing.queuedSendCallback', client.id, null);
        });
        // Clear the active and queued subscriptions lists as we were
        // asked to disconnect.
        logger.log('data', client.id, 'client._subscriptions:',
                   client._subscriptions);
        while (client._subscriptions.length > 0) {
          client._subscriptions.shift();
        }
        while (client._queuedSubscriptions.length > 0) {
          client._queuedSubscriptions.shift();
        }
        // Close our end of the socket
        if (client._stream) {
          client._stream.removeAllListeners('close');
          client._stream.end();
          client._stream = null;
        }
        client._queuedChunks = [];
        client._queuedChunksSize = 0;

        // Indicate that we've disconnected
        client._setState(STATE_STOPPED);
        // Remove ourself from the active client list
        var activeClient = activeClientList.get(client.id);
        if (client === activeClient) activeClientList.remove(client.id);
        setImmediate(function() {
          if (!client._firstStart) {
            client._firstStart = true;
            logger.log('emit', client.id, STATE_STOPPED);
            client.emit(STATE_STOPPED);
          }
        });
        if (callback) {
          process.nextTick(function() {
            logger.entry('Client.stop.performDisconnect.stopProcessing.' +
                'callback', client.id);
            callback.apply(client);
            logger.exit('Client.stop.performDisconnect.stopProcessing.' +
                'callback', client.id, null);
          });
        }

        logger.exit('Client.stop.performDisconnect.stopProcessing',
            client.id, null);
      }, callback);

      logger.exit('Client.stop.performDisconnect', client.id, null);
      return;
    }

    // try disconnect again
    setImmediate(performDisconnect, client, callback);

    logger.exit('Client.stop.performDisconnect', client.id, null);
  }
1221320
function () {
    if (!self.connections.length) {
      self.removeListener('remove', remove);
      self.closing = false;
      return callback();
    }
  }