func (c *Client) makeResponseCallback(doneChan chan *response, removeCallback <-chan uint64, method string, args []interface{}) dnode.Function {
	return dnode.Callback(func(arguments *dnode.Partial) {
		var resp struct {
			Result *dnode.Partial `json:"result"`
			Err    *Error         `json:"error"`
		}
		defer func() {
			if resp.Err != nil {
				c.LocalKite.Log.Debug("Error received from kite: %q method: %q args: %#v err: %s", c.Kite.Name, method, args, resp.Err.Error())
				doneChan <- &response{resp.Result, resp.Err}
			} else {
				doneChan <- &response{resp.Result, nil}
			}
		}()
		if id, ok := <-removeCallback; ok {
			c.scrubber.RemoveCallback(id)
		}
		arg, err := arguments.SliceOfLength(1)
		if err != nil {
			resp.Err = &Error{Type: "invalidResponse", Message: err.Error()}
			return
		}
		err = arg[0].Unmarshal(&resp)
		if err != nil {
			resp.Err = &Error{Type: "invalidResponse", Message: err.Error()}
			return
		}
		keys := make(map[string]interface{})
		err = arg[0].Unmarshal(&keys)
		_, ok1 := keys["result"]
		_, ok2 := keys["error"]
		if !ok1 && !ok2 {
			resp.Err = &Error{
				Type:    "invalidResponse",
				Message: "Server has sent invalid response arguments",
			}
			return
		}
	})
}
makeResponseCallback prepares and returns a callback function sent to the server.
The caller of the Tell() is blocked until the server calls this callback function.
Sets theResponse and notifies the caller by sending to done channel.
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()
}
func (c *Client) callOnConnectHandlers() {
	c.m.RLock()
	defer c.m.RUnlock()
	for _, handler := range c.onConnectHandlers {
		func() {
			defer nopRecover()
			handler()
		}()
	}
}
func (c *Client) callOnDisconnectHandlers() {
	c.m.RLock()
	defer c.m.RUnlock()
	for _, handler := range c.onDisconnectHandlers {
		func() {
			defer nopRecover()
			handler()
		}()
	}
}
func (c *Client) callOnTokenExpireHandlers() {
	c.m.RLock()
	defer c.m.RUnlock()
	for _, handler := range c.onTokenExpireHandlers {
		func() {
			defer nopRecover()
			handler()
		}()
	}
}
func (c *Client) callOnTokenRenewHandlers(token string) {
	c.m.RLock()
	defer c.m.RUnlock()
	for _, handler := range c.onTokenRenewHandlers {
		func() {
			defer nopRecover()
			handler(token)
		}()
	}
}
func (c *Client) wrapMethodArgs(args []interface{}, responseCallback dnode.Function) []interface{} {
	options := callOptionsOut{
		WithArgs: args,
		callOptions: callOptions{
			Kite:             *c.LocalKite.Kite(),
			Auth:             c.authCopy(),
			ResponseCallback: responseCallback,
		},
	}
	return []interface{}{options}
}
func (c *Client) Tell(method string, args ...interface{}) (result *dnode.Partial, err error) {
	return c.TellWithTimeout(method, 0, args...)
}
func (c *Client) SendWebRTCRequest(req *protocol.WebRTCSignalMessage) error {
	timeout := time.Duration(0)
	if c.Config != nil {
		timeout = c.Config.Timeout
	}
	_, err := c.TellWithTimeout(WebRTCHandlerName, timeout, req)
	return err
}
func (c *Client) TellWithTimeout(method string, timeout time.Duration, args ...interface{}) (result *dnode.Partial, err error) {
	response := <-c.GoWithTimeout(method, timeout, args...)
	return response.Result, response.Err
}
func (c *Client) Go(method string, args ...interface{}) chan *response {
	return c.GoWithTimeout(method, 0, args...)
}
func (c *Client) GoWithTimeout(method string, timeout time.Duration, args ...interface{}) chan *response {
	responseChan := make(chan *response, 1)
	c.sendMethod(method, args, timeout, responseChan)
	return responseChan
}
func (c *Client) sendMethod(method string, args []interface{}, timeout time.Duration, responseChan chan *response) {
	removeCallback := make(chan uint64, 1)
	doneChan := make(chan *response, 1)
	cb := c.makeResponseCallback(doneChan, removeCallback, method, args)
	args = c.wrapMethodArgs(args, cb)
	callbacks, errC, err := c.marshalAndSend(method, args)
	if err != nil {
		responseChan <- &response{
			Result: nil,
			Err: &Error{
				Type:    "sendError",
				Message: err.Error(),
			},
		}
		return
	}
	var afterTimeout <-chan time.Time
	if timeout > 0 {
		afterTimeout = time.After(timeout)
	}
	go func() {
		c.disconnectMu.Lock()
		defer c.disconnectMu.Unlock()
		select {
		case resp := <-doneChan:
			if e, ok := resp.Err.(*Error); ok {
				if e.Type == "authenticationError" && strings.Contains(e.Message, "token is expired") {
					c.callOnTokenExpireHandlers()
				}
			}
			responseChan <- resp
		case <-c.disconnect:
			responseChan <- &response{
				nil,
				&Error{
					Type:    "disconnect",
					Message: "Remote kite has disconnected",
				},
			}
		case err := <-errC:
			if err != nil {
				responseChan <- &response{
					nil,
					&Error{
						Type:    "sendError",
						Message: err.Error(),
					},
				}
			}
		case <-afterTimeout:
			responseChan <- &response{
				nil,
				&Error{
					Type:    "timeout",
					Message: fmt.Sprintf("No response to %q method in %s", method, timeout),
				},
			}
			if id, ok := <-removeCallback; ok {
				c.scrubber.RemoveCallback(id)
			}
		}
	}()
	sendCallbackID(callbacks, removeCallback)
}
func (c *Client) marshalAndSend(method interface{}, arguments []interface{}) (callbacks map[string]dnode.Path, errC <-chan error, err error) {
	callbacks = c.scrubber.Scrub(arguments)
	defer func() {
		if err != nil {
			c.removeCallbacks(callbacks)
		}
	}()
	if arguments == nil {
		arguments = make([]interface{}, 0)
	}
	rawArgs, err := json.Marshal(arguments)
	if err != nil {
		return nil, nil, err
	}
	msg := dnode.Message{
		Method:    method,
		Arguments: &dnode.Partial{Raw: rawArgs},
		Callbacks: callbacks,
	}
	p, err := json.Marshal(msg)
	if err != nil {
		return nil, nil, err
	}
	select {
	case <-c.closeChan:
		return nil, nil, errors.New("can't send, client is closed")
	default:
		if c.getSession() == nil {
			return nil, nil, errors.New("can't send, session is not established yet")
		}
		errC := make(chan error, 1)
		c.send <- &message{
			p:    p,
			errC: errC,
		}
		return callbacks, errC, nil
	}
}
func (c *Client) getSession() sockjs.Session {
	c.m.RLock()
	defer c.m.RUnlock()
	return c.session
}
func (c *Client) setSession(session sockjs.Session) {
	c.testHookSetSession(session)
	c.m.Lock()
	c.session = session
	c.m.Unlock()
}
func (c *Client) removeCallbacks(callbacks map[string]dnode.Path) {
	for sid := range callbacks {
		id, _ := strconv.ParseUint(sid, 10, 64)
		c.scrubber.RemoveCallback(id)
	}
}
func (c *Client) config() *config.Config {
	if c.Config != nil {
		return c.Config
	}
	return c.LocalKite.Config
}
func sendCallbackID(callbacks map[string]dnode.Path, ch chan<- uint64) {
	for id, path := range callbacks {
		if len(path) != 2 {
			continue
		}
		p0, ok := path[0].(string)
		if !ok {
			continue
		}
		p1, ok := path[1].(string)
		if !ok {
			continue
		}
		if p0 != "0" || p1 != "responseCallback" {
			continue
		}
		i, _ := strconv.ParseUint(id, 10, 64)
		ch <- i
		return
	}
	close(ch)
}