func main() {

	// Default logger
	logging.Debug("Debug")
	logging.Info("Info")
	logging.Notice("Notice")
	logging.Warning("Warning")
	logging.Error("Error")
	logging.Critical("Critical")

	// Custom logger with default handler
	l := logging.NewLogger("test")

	l.Debug("Debug")
	l.Info("Info")
	l.Notice("Notice")
	l.Warning("Warning")
	l.Error("Error")
	l.Critical("Critical")

	// Custom logger with custom handler
	l2 := logging.NewLogger("test2")
	l2.SetHandler(&MyHandler{})

	l2.Debug("Debug")
	l2.Info("Info")
	l2.Notice("Notice")
	l2.Warning("Warning")
	l2.Error("Error")
	l2.Critical("Critical")
}

func NewRemote(machine string) (*Remote, error) {
	log := logging.NewLogger("fusetest")

	opts := sshCmd.SSHCommandOpts{Ask: true}
	s, err := sshCmd.NewSSHCommand(log, opts)
	if err != nil {
		return nil, err
	}

	info, err := getMachineInfo(s, machine)
	if err != nil {
		return nil, err
	}

	if len(info.Mounts) == 0 {
		return nil, errors.New("Machine has no mount. Please mount and try again.")
	}

	conn, err := dialSSH(s.PrivateKeyPath(), info.Hostname, info.IP)
	if err != nil {
		return nil, err
	}

	return &Remote{
		SSHConn: conn,
		remote:  info.Mounts[0].RemotePath,
		local:   info.Mounts[0].LocalPath,
	}, nil
}

func NewPool(k *kite.Kite) *KlientPool {
	return &KlientPool{
		kite:    k,
		klients: make(map[string]*Klient),
		log:     logging.NewLogger("klientpool"),
	}
}

func newLogger(name string, debug bool) logging.Logger {
	log := logging.NewLogger(name)
	logHandler := logging.NewWriterHandler(os.Stderr)
	logHandler.Colorize = true
	log.SetHandler(logHandler)

	if debug {
		log.SetLevel(logging.DEBUG)
		logHandler.SetLevel(logging.DEBUG)
	}

	return log
}

// ReadDirectoryHandler calls a remote machine's fs.readDirectory method with
// the given args.
func (r *Remote) ReadDirectoryHandler(kreq *kite.Request) (interface{}, error) {
	log := logging.NewLogger("remote").New("remote.readDirectory")

	var params req.ReadDirectoryOptions
	if kreq.Args == nil {
		return nil, errors.New("arguments are not passed")
	}

	if err := kreq.Args.One().Unmarshal(&params); err != nil {
		err = fmt.Errorf(
			"remote.readDirectory: Error '%s' while unmarshalling request '%s'\n",
			err, kreq.Args.One(),
		)
		r.log.Error(err.Error())
		return nil, err
	}

	switch {
	case params.Machine == "":
		return nil, errors.New("Missing required argument `machine`.")
	}

	log = log.New(
		"machineName", params.Machine,
		"remotePath", params.Path,
	)

	remoteMachine, err := r.GetDialedMachine(params.Machine)
	if err != nil {
		log.Error("Error getting dialed, valid machine. err:%s", err)
		return nil, err
	}

	if params.Path == "" {
		home, err := remoteMachine.HomeWithDefault()
		if err != nil {
			return nil, err
		}
		params.Path = home
	}

	if !filepath.IsAbs(params.Path) {
		home, err := remoteMachine.HomeWithDefault()
		if err != nil {
			return nil, err
		}
		params.Path = path.Join(home, params.Path)
	}

	return remoteMachine.Tell("fs.readDirectory", params)
}

// New creates a new Kloud instance without initializing the default providers.
func New() *Kloud {
	log := logging.NewLogger(NAME)

	kld := &Kloud{
		idlock:      idlock.New(),
		Log:         log,
		Eventers:    make(map[string]eventer.Eventer),
		providers:   make(map[string]Provider),
		statusCache: cache.NewMemoryWithTTL(time.Second * 10),
	}

	kld.statusCache.StartGC(time.Second * 5)

	return kld
}

// newLogger returns a new kite logger based on koding/logging package and a
// SetLogLvel function. The current logLevel is INFO by default, which can be
// changed with KITE_LOG_LEVEL environment variable.
func newLogger(name string) (Logger, func(Level)) {
	logger := logging.NewLogger(name)
	logger.SetLevel(convertLevel(getLogLevel()))

	if os.Getenv("KITE_LOG_NOCOLOR") != "" {
		logging.StdoutHandler.Colorize = false
		logging.StderrHandler.Colorize = false
	}

	setLevel := func(l Level) {
		logger.SetLevel(convertLevel(l))
		logging.DefaultHandler.SetLevel(convertLevel(l))
	}

	return logger, setLevel
}

func TestMain(m *testing.M) {
	var err error
	vg, err = NewVagrant(vagrantName)
	if err != nil {
		log.Fatalln(err)
	}

	vg.Log = logging.NewLogger("vagrantutil_test")
	vg.Log.SetLevel(logging.DEBUG)
	h := logging.NewWriterHandler(os.Stderr)
	h.SetLevel(logging.DEBUG)
	vg.Log.SetHandler(h)

	ret := m.Run()
	os.RemoveAll(vagrantName)
	os.Exit(ret)
}

func main() {
	rmq := rabbitmq.New(
		&rabbitmq.Config{
			Host:     "localhost",
			Port:     5672,
			Username: "guest",
			Password: "guest",
			Vhost:    "/",
		},
		logging.NewLogger("producer"),
	)

	exchange := rabbitmq.Exchange{
		Name: "EXCHANGE_NAME",
	}

	queue := rabbitmq.Queue{}
	publishingOptions := rabbitmq.PublishingOptions{
		Tag:        "ProducerTagHede",
		RoutingKey: "naber",
	}

	publisher, err := rmq.NewProducer(exchange, queue, publishingOptions)
	if err != nil {
		panic(err)
	}
	defer publisher.Shutdown()
	publisher.RegisterSignalHandler()

	// may be we should autoconvert to byte array?
	msg := amqp.Publishing{
		Body: []byte("2"),
	}

	publisher.NotifyReturn(func(message amqp.Return) {
		fmt.Println(message)
	})

	for i := 0; i < 10; i++ {
		err = publisher.Publish(msg)
		if err != nil {
			fmt.Println(err, i)
		}
	}
}

func main() {
	// command line parameter parsing
	configfile := flag.String("cfg", "mambo.cfg", "Main configuration file")
	flag.Parse()
	logger := logging.NewLogger("Mambo")
	logger.Notice("Mambo collector started")
	logger.Notice("Loading configuration from %s", *configfile)
	// The 'results' channel will recive the results of the mysqlWorker queries
	results := make(chan string)
	config, commands := configure(*configfile) // Loading configuration and commands from ini file
	for _, command := range commands {
		go controller(command, config, results) // every command will launch a command controller
	}
	logger.Notice("Data collector running")
	for {
		select {
		// every time a MySQL worker yield data to the 'results' channel we call a statsdSender and we send that data to statsdserver
		case msg := <-results:
			{
				statsdSender(config, msg)
			}
		}
	}
}

// NewRemote creates a new Remote instance.
func NewRemote(opts *RemoteOptions) *Remote {
	// TODO: Improve this usage. Basically i want a proper koding/logging struct,
	// but we need to create it from somewhere. klient is always uses kite.Logger,
	// which **should** always be implemented by a koding/logging.Logger.. but in
	// the event that it's not, how do we handle it?
	kodingLog, ok := opts.Log.(logging.Logger)
	if !ok {
		opts.Log.Error(
			"Unable to convert koding/kite.Logger to koding/logging.Logger. Creating new logger",
		)
		kodingLog = logging.NewLogger("new-logger")
	}

	kodingLog = kodingLog.New("remote")
	// Setting the handler to debug, because various Remote methods allow a
	// debug option, and this saves us from having to set the handler level every time.
	// This only sets handler, not the actual loglevel.
	logging.DefaultHandler.SetLevel(logging.DEBUG)

	r := &Remote{
		localKite:            opts.Kite,
		kitesGetter:          &KodingKite{Kite: opts.Kite},
		storage:              opts.Storage,
		log:                  kodingLog,
		machinesErrCacheMax:  5 * time.Minute,
		machinesCacheMax:     10 * time.Second,
		machineNamesCache:    map[string]string{},
		unmountPath:          fuseklient.Unmount,
		machines:             machine.NewMachines(kodingLog, opts.Storage),
		maxRestoreAttempts:   defaultMaxRestoreAttempts,
		restoreFailuresPause: defaultRestoreFailuresPause,
		eventSub:             opts.EventSub,
	}

	return r
}

	"strings"
	"time"

	"koding/kites/kloud/api/amazon"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/service/route53"
	"github.com/cenkalti/backoff"
	"github.com/dchest/validator"
	"github.com/koding/logging"
)

var ErrNoRecord = errors.New("no records available")

var defaultLog = logging.NewLogger("kloud-dns")

type Route53 struct {
	*route53.Route53
	ZoneId string

	opts *Options
}

type Options struct {
	Creds      *credentials.Credentials
	HostedZone string
	Log        logging.Logger

	// SyncTimeout tells at most how much time we're going to wait
	// till DNS change is propageted on all Amazon servers.

package lookup

import (
	"koding/kites/kloud/api/amazon"

	"github.com/koding/logging"
)

var defaultLogger = logging.NewLogger("lookup")

type Lookup struct {
	// values contains a list of instance tags that are identified as test
	// instances. By default all instances are fetched.
	values  []string
	clients *amazon.Clients
	log     logging.Logger
}

// NewAWS gives new Lookup client.
//
// When opts.Log is nil, defaultLogger is used instead.
func NewAWS(opts *amazon.ClientOptions) (*Lookup, error) {
	optsCopy := *opts
	if optsCopy.Log == nil {
		optsCopy.Log = defaultLogger
	}
	clients, err := amazon.NewClients(&optsCopy)
	if err != nil {
		return nil, err
	}
	return &Lookup{

// GetPathSize gets the size of a remote path.
func (r *Remote) GetPathSize(kreq *kite.Request) (interface{}, error) {
	log := logging.NewLogger("remote").New("remote.getPathSize")

	var opts req.GetPathSizeOptions
	if kreq.Args == nil {
		return nil, errors.New("arguments are not passed")
	}

	if err := kreq.Args.One().Unmarshal(&opts); err != nil {
		err = fmt.Errorf(
			"remote.getPathSize: %s, while unmarshalling request: %q\n",
			err, kreq.Args.One(),
		)
		log.Error(err.Error())
		return nil, err
	}

	if opts.Machine == "" {
		return nil, errors.New("missing required argument: machine")
	}

	if opts.Debug {
		log.SetLevel(logging.DEBUG)
	}

	log = log.New(
		"machineName", opts.Machine,
		"remotePath", opts.RemotePath,
	)

	remoteMachine, err := r.GetDialedMachine(opts.Machine)
	if err != nil {
		log.Error("Error getting dialed, valid machine. err:%s", err)
		return nil, err
	}

	if opts.RemotePath == "" {
		log.Debug("RemotePath option is empty, finding default.")

		home, err := remoteMachine.HomeWithDefault()
		if err != nil {
			log.Error("Failed to get remote home. err:%s", err)
			return nil, err
		}
		opts.RemotePath = home
	}

	if !filepath.IsAbs(opts.RemotePath) {
		log.Debug("RemotePath is not absolute. Joining it to home.")

		home, err := remoteMachine.HomeWithDefault()
		if err != nil {
			log.Error("Failed to get remote home. err:%s", err)
			return nil, err
		}
		opts.RemotePath = path.Join(home, opts.RemotePath)
	}

	exists, err := remoteMachine.DoesRemotePathExist(opts.RemotePath)
	if err != nil {
		log.Error("Unable to determine if remote path exists. err:%s", err)
		return nil, err
	}

	if !exists {
		log.Error("Remote path does not exist. path:%s", opts.RemotePath)
		return nil, mount.ErrRemotePathDoesNotExist
	}

	size, err := remoteMachine.GetFolderSize(opts.RemotePath)
	if err != nil {
		log.Error("Failed to get remote size. path:%s, err:%s", opts.RemotePath, err)
	}

	return size, err
}

func realMain() error {
	conf := new(Config)
	multiconfig.MustLoadWithPath("config.toml", conf)

	db := mongodb.NewMongoDB(conf.MongoURL)
	opts := &amazon.ClientOptions{
		Credentials: credentials.NewStaticCredentials(conf.AccessKey, conf.SecretKey, ""),
		Regions:     amazon.ProductionRegions,
		Log:         logging.NewLogger("userdebug"),
	}
	ec2clients, err := amazon.NewClients(opts)
	if err != nil {
		panic(err)
	}

	w := new(tabwriter.Writer)
	w.Init(os.Stdout, 10, 8, 0, '\t', 0)
	defer w.Flush()

	// search via username, mongodb -> aws
	if conf.Username != "" {
		ms, err := machinesFromUsername(db, conf.Username)
		if err == nil {
			ms.Print(w)

			for _, m := range ms.docs {
				// if the mongodb document has a region and instanceId, display it too
				if m.Meta.Region != "" && m.Meta.InstanceId != "" {
					client, err := ec2clients.Region(m.Meta.Region)
					if err != nil {
						return err
					}

					i, err := awsData(client, m.Meta.InstanceId)
					if err != nil {
						return err
					}

					i.Print(w)
				}
			}
		} else {
			fmt.Fprintf(os.Stderr, err.Error())
		}
	}

	// search via instanceId, aws -> mongodb
	if conf.InstanceId != "" {
		for _, client := range ec2clients.Regions() {
			i, err := awsData(client, conf.InstanceId)
			if err != nil {
				continue // if not found continue with next region
			}

			// if we find a mongoDB document display it
			ms, err := machinesFromInstanceId(db, conf.InstanceId)
			if err == nil {
				ms.Print(w)
			}

			i.Print(w)
			break
		}
	}

	return nil
}

func TestPublish(t *testing.T) {
	ps := NewPubSub(logging.NewLogger("testing"))
	s := kite.New("s", "0.0.0")
	s.Config.DisableAuthentication = true

	doneC, publish := handlerWrapper(ps.Publish)
	s.HandleFunc("client.Publish", publish)

	ts := httptest.NewServer(s)
	defer ts.Close()

	k := kite.New("c", "0.0.0")
	c := k.NewClient(fmt.Sprintf("%s/kite", ts.URL))

	err := c.Dial()
	if err != nil {
		t.Fatal("Failed to connect to testing Kite", err)
	}

	// Should require args
	_, err = c.Tell("client.Publish")
	if err == nil {
		t.Error("client.Publish should require args")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should require eventName
	_, err = c.Tell("client.Publish", struct {
		Random string
		Data   string
	}{
		Random: "foo",
		Data:   "bar",
	})
	if err == nil {
		t.Error("client.Publish should require EventName")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should require subscriptions for the given event
	_, err = c.Tell("client.Publish", PublishRequest{
		EventName: "foo",
	})
	if err == nil {
		t.Error("client.Publish should return an error, without any subs")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should call onPublish callbacks
	callbackCount := 0
	ps.Subscriptions["test"] = map[int]dnode.Function{
		0: {mockCaller(func(v ...interface{}) error {
			callbackCount += 1
			return nil
		})},
		1: {mockCaller(func(v ...interface{}) error {
			callbackCount += 2
			return nil
		})},
	}

	_, err = c.Tell("client.Publish", PublishRequest{
		EventName: "test",
	})
	if err != nil {
		t.Fatal("client.Publish should call onPublish callbacks without error.", err)
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	if callbackCount != 3 {
		t.Fatal("client.Publish should call onPublish callbacks")
	}

	// Should publish arbitrary data
	var b []byte
	updatedC := make(chan struct{}, 1)
	ps.Subscriptions["other"] = map[int]dnode.Function{
		0: {mockCaller(func(v ...interface{}) error {
			b = v[0].([]interface{})[0].(*dnode.Partial).Raw
			select {
			case updatedC <- struct{}{}:
			case <-time.After(time.Second):
			}
			return nil
		})},
	}

	_, err = c.Tell("client.Publish", struct {
		EventName string
		CountData int
		ListData  []string
	}{
//.........这里部分代码省略.........

	"flag"         // Command line parsing
	"fmt"          // Output formatting

	_ "github.com/go-sql-driver/mysql" // MySQL connection
	"os"                               // to exit with exitcode
	"strconv"                          // string conversion
	"strings"                          // string manipulation
	"time"                             // timestamp logging, ticker

	"github.com/cactus/go-statsd-client/statsd" // Statsd client
	_ "github.com/go-sql-driver/mysql"          // MySQL connection
	"github.com/koding/logging"                 // logging
	"gopkg.in/ini.v1"                           // ini file parsing
)

var logger = logging.NewLogger("Mambo")

/*
  Configuration parameters, mysql & statsd
*/
type configuration struct {
	mysqlHost  string // MySQL host to connect, if empty local socket will be used
	mysqlUser  string // User to connect MySQL with
	mysqlPass  string // Password for connecting MySQL
	mysqlDb    string // Database to connect to
	mysqlPort  int    // Port to connect MySQL, if left blank, 3306 will be used as default
	statsdHost string // statsd server hostname
	statsdPort int    // statsd server port, if left blank, 8125 will be used as default
}

/*

func TestUnsubscribe(t *testing.T) {
	ps := NewPubSub(logging.NewLogger("testing"))
	s := kite.New("s", "0.0.0")
	s.Config.DisableAuthentication = true

	donePubC, publish := handlerWrapper(ps.Publish)
	s.HandleFunc("client.Publish", publish)

	doneSubC, subscribe := handlerWrapper(ps.Subscribe)
	s.HandleFunc("client.Subscribe", subscribe)

	doneUnsubC, unsubscribe := handlerWrapper(ps.Unsubscribe)
	s.HandleFunc("client.Unsubscribe", unsubscribe)

	ts := httptest.NewServer(s)
	defer ts.Close()

	c1 := kite.New("c1", "0.0.0").NewClient(fmt.Sprintf("%s/kite", ts.URL))
	c2 := kite.New("c2", "0.0.0").NewClient(fmt.Sprintf("%s/kite", ts.URL))

	err := c1.Dial()
	if err != nil {
		t.Fatal("Failed to connect to testing Kite", err)
	}
	err = c2.Dial()
	if err != nil {
		t.Fatal("Failed to connect to testing Kite", err)
	}

	// Track the calls to our subs.
	callsMu := sync.Mutex{} // protects calls map.
	calls := map[string]bool{}
	var wg sync.WaitGroup
	wg.Add(3)

	// Setup our event, sub index 1
	_, err = c1.Tell("client.Subscribe", SubscribeRequest{
		EventName: "test",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {
			defer wg.Done()

			callsMu.Lock()
			defer callsMu.Unlock()
			calls["c1:1"] = true
		}),
	})
	if err != nil {
		t.Fatal(err)
	}
	if err = wait(doneSubC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Setup our event, sub index 2
	_, err = c2.Tell("client.Subscribe", SubscribeRequest{
		EventName: "test",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {
			defer wg.Done()

			callsMu.Lock()
			defer callsMu.Unlock()
			calls["c2:2"] = true
		}),
	})
	if err != nil {
		t.Fatal(err)
	}
	if err = wait(doneSubC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Setup our event, sub index 3
	_, err = c2.Tell("client.Subscribe", SubscribeRequest{
		EventName: "test",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {
			defer wg.Done()

			callsMu.Lock()
			defer callsMu.Unlock()
			calls["c2:3"] = true
		}),
	})
	if err != nil {
		t.Fatal(err)
	}
	if err = wait(doneSubC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Setup our event, sub index 4
	_, err = c1.Tell("client.Subscribe", SubscribeRequest{
		EventName: "test",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {
			defer wg.Done()

			callsMu.Lock()
			defer callsMu.Unlock()
			calls["c1:4"] = true
		}),
	})
//.........这里部分代码省略.........

	mysqlport   int    // Port to connect MySQL, if left blank, 3306 will be used as default
	binlogdir   string // Directory to keep binlogs
	mysqlbinlog string // mysqlbinlog binary with full path
	daysKeep    int64  // days to keep binlogs
}

type Binlog struct {
	filename string
	filesize int64
}

var (
	remoteBinlogs  []Binlog
	localBinlogs   []Binlog
	missingBinlogs []Binlog
	logger         = logging.NewLogger("binlogstreamer")
)

func main() {
	configfile := flag.String("cfg", "binlogstreamer.cfg", "Configuration file")
	flag.Parse()
	logger.Notice("Binlogstreamer started")
	logger.Notice("Loading configuration from %s", *configfile)
	config := configure(*configfile)
	remoteBinlogs := getRemoteBinlogs(config)
	localBinlogs := getLocalBinlogs(config)
	missingBinlogs := checkMissingBinlogs(config, localBinlogs, remoteBinlogs)
	go streamBinlogs(config, missingBinlogs)
	cleanupBinlogs(config)
	tick := time.NewTicker(time.Millisecond * 600000).C
	for {

func TestSubscribe(t *testing.T) {
	ps := NewPubSub(logging.NewLogger("testing"))

	s := kite.New("s", "0.0.0")
	s.Config.DisableAuthentication = true

	doneC, subscribe := handlerWrapper(ps.Subscribe)
	s.HandleFunc("client.Subscribe", subscribe)

	ts := httptest.NewServer(s)
	defer ts.Close()

	c1 := kite.New("c1", "0.0.0").NewClient(fmt.Sprintf("%s/kite", ts.URL))
	c2 := kite.New("c2", "0.0.0").NewClient(fmt.Sprintf("%s/kite", ts.URL))

	err := c1.Dial()
	if err != nil {
		t.Fatal("Failed to connect to testing Kite", err)
	}
	err = c2.Dial()
	if err != nil {
		t.Fatal("Failed to connect to testing Kite", err)
	}

	// Should require arguments
	_, err = c1.Tell("client.Subscribe")
	if err == nil {
		t.Error("client.Subscribe should require args")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should require eventName
	_, err = c1.Tell("client.Subscribe", struct {
		Data      string
		OnPublish dnode.Function
	}{
		Data:      "foo",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {}),
	})
	if err == nil {
		t.Error("client.Subscribe should require EventName")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should require onPublish
	_, err = c1.Tell("client.Subscribe", struct {
		eventName string
		Data      string
	}{
		eventName: "foo",
		Data:      "bar",
	})
	if err == nil {
		t.Error("client.Subscribe should require OnPublish")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should require valid onPublish func
	_, err = c1.Tell("client.Subscribe", struct {
		eventName string
		onPublish string
	}{
		eventName: "foo",
		onPublish: "bar",
	})
	if err == nil {
		t.Error("client.Subscribe should require a valid OnPublish func")
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	// Should subscribe to any given event name
	pRes, err := c1.Tell("client.Subscribe", SubscribeRequest{
		EventName: "test",
		OnPublish: dnode.Callback(func(f *dnode.Partial) {}),
	})
	if err != nil {
		t.Error(err)
	}
	if err = wait(doneC, time.Second); err != nil {
		t.Fatalf("want err = nil; got %v", err)
	}

	subs := getCopy(ps, "test")
	if len(subs) != 1 {
		t.Fatal("client.Subscribe should store a single onPublish callback")
	}

	// Should return the subIndex
	var res SubscribeResponse
	if err = pRes.Unmarshal(&res); err != nil {
		t.Errorf("client.Subscribe should return a valid response struct. err:%s", err)
	}
//.........这里部分代码省略.........