func (c *Client) callOnConnectHandlers() {
	c.m.RLock()
	defer c.m.RUnlock()
	for _, handler := range c.onConnectHandlers {
		func() {
			defer nopRecover()
			handler()
		}()
	}
}
callOnConnectHandlers runs the registered connect handlers.
package kite
import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/koding/kite/config"
	"github.com/koding/kite/dnode"
	"github.com/koding/kite/protocol"
	"github.com/koding/kite/sockjsclient"
	"github.com/cenkalti/backoff"
	"github.com/gorilla/websocket"
	"github.com/igm/sockjs-go/sockjs"
)
var forever backoff.BackOff
func init() {
	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = 365 * 24 * time.Hour  
	forever = &lockedBackoff{b: b}
}
func nopSetSession(sockjs.Session) {}
type Client struct {
	protocol.Kite  
	LocalKite *Kite
	Auth *Auth
	Reconnect bool
	URL string
	Config *config.Config
	Concurrent bool
	ConcurrentCallbacks bool
	ClientFunc func(*sockjsclient.DialOptions) *http.Client
	ReadBufferSize int
	WriteBufferSize int
	muProt sync.Mutex  
	disconnect   chan struct{}
	disconnectMu sync.Mutex  
	authMu sync.Mutex
	closeChan chan struct{}
	closeRenewer chan struct{}
	interrupt chan error
	wg sync.WaitGroup
	session sockjs.Session
	send    chan *message
	ctxMu  sync.Mutex
	ctx    context.Context
	cancel func()
	muReconnect sync.Mutex
	closed int32
	scrubber *dnode.Scrubber
	redialBackOff backoff.BackOff
	onConnectHandlers     []func()
	onDisconnectHandlers  []func()
	onTokenExpireHandlers []func()
	onTokenRenewHandlers  []func(string)
	testHookSetSession func(sockjs.Session)
	m sync.RWMutex
	firstRequestHandlersNotified sync.Once
}
type message struct {
	p    []byte
	errC chan<- error
}
type callOptions struct {
	Kite             protocol.Kite  `json:"kite" dnode:"-"`
	Auth             *Auth          `json:"authentication"`
	WithArgs         *dnode.Partial `json:"withArgs" dnode:"-"`
	ResponseCallback dnode.Function `json:"responseCallback"`
}
type callOptionsOut struct {
	callOptions
	WithArgs []interface{} `json:"withArgs"`
}
type Auth struct {
	Type string `json:"type"`
	Key  string `json:"key"`
}
type response struct {
	Result *dnode.Partial
	Err    error
}
func (k *Kite) NewClient(remoteURL string) *Client {
	c := &Client{
		LocalKite:          k,
		URL:                remoteURL,
		disconnect:         make(chan struct{}),
		closeChan:          make(chan struct{}),
		redialBackOff:      forever,
		scrubber:           dnode.NewScrubber(),
		testHookSetSession: nopSetSession,
		Concurrent:         true,
		send:               make(chan *message),
		interrupt:          make(chan error, 1),
		ctx:                context.Background(),
		cancel:             func() {},
	}
	c.OnConnect(c.setContext)
	c.OnDisconnect(c.closeContext)
	k.OnRegister(c.updateAuth)
	return c
}
func (c *Client) SetUsername(username string) {
	c.muProt.Lock()
	c.Kite.Username = username
	c.muProt.Unlock()
}
func (c *Client) Dial() (err error) {
	return c.DialTimeout(0)
}
func (c *Client) DialTimeout(timeout time.Duration) error {
	err := c.dial(timeout)
	c.LocalKite.Log.Debug("Dialing '%s' kite: %s (error: %v)", c.Kite.Name, c.URL, err)
	if err != nil {
		return err
	}
	go c.run()
	return nil
}
func (c *Client) DialForever() (connected chan bool, err error) {
	c.Reconnect = true
	connected = make(chan bool, 1)  
	go c.dialForever(connected)
	return
}
func (c *Client) updateAuth(reg *protocol.RegisterResult) {
	c.authMu.Lock()
	defer c.authMu.Unlock()
	if c.Auth == nil {
		return
	}
	if c.Auth.Type == "kiteKey" && reg.KiteKey != "" {
		c.Auth.Key = reg.KiteKey
	}
}
func (c *Client) setContext() {
	c.ctxMu.Lock()
	c.ctx, c.cancel = context.WithCancel(context.Background())
	c.ctxMu.Unlock()
}
func (c *Client) closeContext() {
	c.ctxMu.Lock()
	c.cancel()
	c.ctxMu.Unlock()
}
func (c *Client) context() context.Context {
	c.ctxMu.Lock()
	defer c.ctxMu.Unlock()
	return c.ctx
}
func (c *Client) authCopy() *Auth {
	c.authMu.Lock()
	defer c.authMu.Unlock()
	if c.Auth == nil {
		return nil
	}
	authCopy := *c.Auth
	return &authCopy
}
func (c *Client) dial(timeout time.Duration) (err error) {
	transport := c.config().Transport
	c.LocalKite.Log.Debug("Client transport is set to '%s'", transport)
	var session sockjs.Session
	switch transport {
	case config.WebSocket:
		session, err = sockjsclient.DialWebsocket(c.URL, c.config())
	case config.XHRPolling:
		session, err = sockjsclient.DialXHR(c.URL, c.config())
	case config.Auto:
		session, err = sockjsclient.DialWebsocket(c.URL, c.config())
		if err == websocket.ErrBadHandshake {
			session, err = sockjsclient.DialXHR(c.URL, c.config())
		}
	default:
		return fmt.Errorf("Connection transport is not known '%v'", transport)
	}
	if err != nil {
		return err
	}
	c.setSession(session)
	c.wg.Add(1)
	go c.sendHub()
	c.redialBackOff.Reset()
	go c.callOnConnectHandlers()
	return nil
}
func (c *Client) dialForever(connectNotifyChan chan bool) {
	dial := func() error {
		if !c.reconnect() {
			return nil
		}
		c.LocalKite.Log.Info("Dialing '%s' kite: %s", c.Kite.Name, c.URL)
		if err := c.dial(0); err != nil {
			c.LocalKite.Log.Warning("Dialing '%s' kite error: %s: %v", c.Kite.Name, c.URL, err)
			return err
		}
		return nil
	}
	backoff.Retry(dial, c.redialBackOff)  
	if connectNotifyChan != nil {
		close(connectNotifyChan)
	}
	go c.run()
}
func (c *Client) RemoteAddr() string {
	session := c.getSession()
	if session == nil {
		return ""
	}
	websocketsession, ok := session.(*sockjsclient.WebsocketSession)
	if !ok {
		return ""
	}
	return websocketsession.RemoteAddr()
}
func (c *Client) run() {
	err := c.readLoop()
	if err != nil {
		c.LocalKite.Log.Debug("readloop err: %s", err)
	}
	c.callOnDisconnectHandlers()
	c.disconnectMu.Lock()
	if c.disconnect != nil {
		close(c.disconnect)
		c.disconnect = nil
	}
	c.disconnectMu.Unlock()
	if c.reconnect() {
		c.disconnectMu.Lock()
		c.disconnect = make(chan struct{}, 1)
		c.disconnectMu.Unlock()
		go c.dialForever(nil)
	}
}
func (c *Client) reconnect() bool {
	c.muReconnect.Lock()
	defer c.muReconnect.Unlock()
	return c.Reconnect
}
func (c *Client) readLoop() error {
	for {
		p, err := c.receiveData()
		c.LocalKite.Log.Debug("readloop received: %s %v", p, err)
		if err != nil {
			return err
		}
		msg, fn, err := c.processMessage(p)
		if err != nil {
			if _, ok := err.(dnode.CallbackNotFoundError); !ok {
				c.LocalKite.Log.Warning("error processing message err: %s message: %s", err, msg)
			}
		}
		switch v := fn.(type) {
		case *Method:  
			if c.Concurrent {
				go c.runMethod(v, msg.Arguments)
			} else {
				c.runMethod(v, msg.Arguments)
			}
		case func(*dnode.Partial):  
			if c.Concurrent && c.ConcurrentCallbacks {
				go c.runCallback(v, msg.Arguments)
			} else {
				c.runCallback(v, msg.Arguments)
			}
		}
	}
}
func (c *Client) receiveData() ([]byte, error) {
	type recv struct {
		msg []byte
		err error
	}
	session := c.getSession()
	if session == nil {
		return nil, errors.New("not connected")
	}
	done := make(chan recv, 1)
	go func() {
		msg, err := session.Recv()
		done <- recv{[]byte(msg), err}
	}()
	select {
	case r := <-done:
		return r.msg, r.err
	case err := <-c.interrupt:
		return nil, err
	}
}
func (c *Client) processMessage(data []byte) (msg *dnode.Message, fn interface{}, err error) {
	defer func() {
		if err != nil {
			onError(err)
		}
	}()
	msg = &dnode.Message{}
	if err = json.Unmarshal(data, &msg); err != nil {
		return nil, nil, err
	}
	sender := func(id uint64, args []interface{}) error {
		_, _, e := c.marshalAndSend(id, args)
		return e
	}
	if err := dnode.ParseCallbacks(msg, sender); err != nil {
		return nil, nil, err
	}
	switch method := msg.Method.(type) {
	case float64:
		id := uint64(method)
		callback := c.scrubber.GetCallback(id)
		if callback == nil {
			err = dnode.CallbackNotFoundError{
				ID:   id,
				Args: msg.Arguments,
			}
			return nil, nil, err
		}
		return msg, callback, nil
	case string:
		m, ok := c.LocalKite.handlers[method]
		if !ok {
			err = dnode.MethodNotFoundError{
				Method: method,
				Args:   msg.Arguments,
			}
			return nil, nil, err
		}
		return msg, m, nil
	default:
		return nil, nil, fmt.Errorf("Method is not string or integer: %+v (%T)", msg.Method, msg.Method)
	}
}
func (c *Client) Close() {
	if !atomic.CompareAndSwapInt32(&c.closed, 0, 1) {
		return  
	}
	c.muReconnect.Lock()
	c.Reconnect = false
	c.muReconnect.Unlock()
	close(c.closeChan)
	if c.closeRenewer != nil {
		select {
		case c.closeRenewer <- struct{}{}:
		default:
		}
	}
	c.wg.Wait()
	if session := c.getSession(); session != nil {
		session.Close(3000, "Go away!")
	}
}
func (c *Client) sendHub() {
	defer c.wg.Done()
	for {
		select {
		case msg := <-c.send:
			c.LocalKite.Log.Debug("sending: %s", msg)
			session := c.getSession()
			if session == nil {
				c.LocalKite.Log.Error("not connected")
				continue
			}
			err := session.Send(string(msg.p))
			if err != nil {
				if msg.errC != nil {
					msg.errC <- err
				}
				if sockjsclient.IsSessionClosed(err) {
					select {
					case c.interrupt <- err:
					default:
					}
					c.LocalKite.Log.Error("error sending to %s: %s", session.ID(), err)
					return
				}
			}
		case <-c.closeChan:
			c.LocalKite.Log.Debug("Send hub is closed")
			return
		}
	}
}
func (c *Client) OnConnect(handler func()) {
	c.m.Lock()
	c.onConnectHandlers = append(c.onConnectHandlers, handler)
	c.m.Unlock()
}
func (c *Client) OnDisconnect(handler func()) {
	c.m.Lock()
	c.onDisconnectHandlers = append(c.onDisconnectHandlers, handler)
	c.m.Unlock()
}
func (c *Client) OnTokenExpire(handler func()) {
	c.m.Lock()
	c.onTokenExpireHandlers = append(c.onTokenExpireHandlers, handler)
	c.m.Unlock()
}
func (c *Client) OnTokenRenew(handler func(token string)) {
	c.m.Lock()
	c.onTokenRenewHandlers = append(c.onTokenRenewHandlers, handler)
	c.m.Unlock()
}