func (c *Client) initTopics() {
	c.psmux.Lock()
	for name := range c.topicHandlerMap {
		topicName := name
		go util.Safe(func() {
			for i := 0; i < 10; i++ {
				topic, _ := newTopic(topicName, util.ValueToBytes(c.Codec, nil))
				bs, _ := topic.toBytes()
				err := c.Call(routeSubscribe, bs, nil, time.Second*10)
				if err == nil {
					log.Info("%v [Subscribe] [topic: '%v'] success from\t%v", c.Handler.LogTag(), topicName, c.Conn.RemoteAddr())
					break
				} else {
					log.Error("%v [Subscribe] [topic: '%v'] %v times failed: %v, from\t%v", c.Handler.LogTag(), topicName, i+1, err, c.Conn.RemoteAddr())
				}
				time.Sleep(time.Second)
			}
		})
	}
	c.psmux.Unlock()
}
func (c *Client) invalidTopic(topic string) bool {
c.psmux.Lock()
_, ok := c.topicHandlerMap[topic]
c.psmux.Unlock()
return !ok
}
package pubsub
import (
	"net"
	"sync"
	"time"
	"github.com/lesismal/arpc"
	"github.com/lesismal/arpc/log"
	"github.com/lesismal/arpc/util"
)
type Client struct {
	*arpc.Client
	Password string
	psmux sync.Mutex
	topicHandlerMap map[string]TopicHandler
	onPublishHandler TopicHandler
}
func (c *Client) Authenticate() error {
	err := c.Call(routeAuthenticate, c.Password, nil, time.Second*5)
	if err == nil {
		log.Info("%v [Authenticate] success from\t%v", c.Handler.LogTag(), c.Conn.RemoteAddr())
	} else {
		log.Error("%v [Authenticate] failed: %v, from\t%v", c.Handler.LogTag(), err, c.Conn.RemoteAddr())
	}
	return err
}
func (c *Client) Subscribe(topicName string, h TopicHandler, timeout time.Duration) error {
	topic, err := newTopic(topicName, nil)
	if err != nil {
		return err
	}
	bs, err := topic.toBytes()
	if err != nil {
		return err
	}
	c.psmux.Lock()
	c.topicHandlerMap[topicName] = h
	c.psmux.Unlock()
	err = c.Call(routeSubscribe, bs, nil, timeout)
	if err == nil {
		log.Info("%v [Subscribe] [topic: '%v'] success from\t%v", c.Handler.LogTag(), topicName, c.Conn.RemoteAddr())
	} else {
		c.psmux.Lock()
		delete(c.topicHandlerMap, topicName)
		c.psmux.Unlock()
		log.Error("%v [Subscribe] [topic: '%v'] failed: %v, from\t%v", c.Handler.LogTag(), topicName, err, c.Conn.RemoteAddr())
	}
	return err
}
func (c *Client) Unsubscribe(topicName string, timeout time.Duration) error {
	topic, err := newTopic(topicName, nil)
	if err != nil {
		return err
	}
	bs, err := topic.toBytes()
	if err != nil {
		return err
	}
	err = c.Call(routeUnsubscribe, bs, nil, timeout)
	if err == nil {
		c.psmux.Lock()
		delete(c.topicHandlerMap, topic.Name)
		c.psmux.Unlock()
		log.Info("%v[Unsubscribe] [topic: '%v'] success from\t%v", c.Handler.LogTag(), topicName, c.Conn.RemoteAddr())
	} else {
		log.Error("%v[Unsubscribe] [topic: '%v'] failed: %v, from\t%v", c.Handler.LogTag(), topicName, err, c.Conn.RemoteAddr())
	}
	return err
}
func (c *Client) Publish(topicName string, v interface{}, timeout time.Duration) error {
	topic, err := newTopic(topicName, util.ValueToBytes(c.Codec, v))
	if err != nil {
		return err
	}
	bs, err := topic.toBytes()
	if err != nil {
		return err
	}
	err = c.Call(routePublish, bs, nil, timeout)
	if err != nil {
		log.Error("%v [Publish] [topic: '%v'] failed: %v, from\t%v", c.Handler.LogTag(), topicName, err, c.Conn.RemoteAddr())
	}
	return err
}
func (c *Client) PublishToOne(topicName string, v interface{}, timeout time.Duration) error {
	topic, err := newTopic(topicName, util.ValueToBytes(c.Codec, v))
	if err != nil {
		return err
	}
	bs, err := topic.toBytes()
	if err != nil {
		return err
	}
	err = c.Call(routePublishToOne, bs, nil, timeout)
	if err != nil {
		log.Error("%v [PublishToOne] [topic: '%v'] failed: %v, from\t%v", c.Handler.LogTag(), topicName, err, c.Conn.RemoteAddr())
	}
	return err
}
func (c *Client) OnPublish(h TopicHandler) {
	c.onPublishHandler = h
}