func (p *PubSub) RemoveTopic(topic string) {
	p.topicMutex.Lock()
	defer p.topicMutex.Unlock()
	info, ok := p.topics[topic]
	if ok {
		log.Info("RemoveTopic", "topic", topic)
		info.cancel()
		info.sub.Cancel()
		err := info.pubtopic.Close()
		if err != nil {
			log.Error("RemoveTopic", "topic", topic, "close topic err", err)
		}
		delete(p.topics, topic)
	}
}
RemoveTopic remove topic
package extension
import (
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/33cn/chain33/types"
	"github.com/33cn/chain33/common/log/log15"
	p2ptypes "github.com/33cn/chain33/system/p2p/dht/types"
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/peer"
	pubsub "github.com/libp2p/go-libp2p-pubsub"
)
var log = log15.New("module", "pubsub")
var setOnce sync.Once
type TopicMap map[string]*topicinfo
type topicinfo struct {
	pubtopic *pubsub.Topic
	sub      *pubsub.Subscription
	ctx      context.Context
	cancel   context.CancelFunc
	topic    string
}
type PubSub struct {
	*pubsub.PubSub
	topics     TopicMap
	topicMutex sync.RWMutex
	ctx        context.Context
}
type SubMsg *pubsub.Message
type SubCallBack func(topic string, msg SubMsg)
func setPubSubParameters(psConf *p2ptypes.PubSubConfig) {
	psConf.MaxMsgSize *= 1 << 20
	if psConf.MaxMsgSize <= 0 {
		psConf.MaxMsgSize = types.MaxBlockSize
	}
	if psConf.PeerOutboundQueueSize <= 0 {
		psConf.PeerOutboundQueueSize = 128
	}
	if psConf.ValidateQueueSize <= 0 {
		psConf.ValidateQueueSize = 128
	}
	if psConf.GossipSubDlo <= 0 {
		psConf.GossipSubDlo = 6
	}
	if psConf.GossipSubD <= 0 {
		psConf.GossipSubD = 8
	}
	if psConf.GossipSubDhi <= 0 {
		psConf.GossipSubDhi = 15
	}
	if psConf.GossipSubHeartbeatInterval <= 0 {
		psConf.GossipSubHeartbeatInterval = 700
	}
	if psConf.GossipSubHistoryGossip <= 0 {
		psConf.GossipSubHistoryGossip = 3
	}
	if psConf.GossipSubHistoryLength <= 0 {
		psConf.GossipSubHistoryLength = 10
	}
	if psConf.GossipSubDlo > psConf.GossipSubD || psConf.GossipSubD > psConf.GossipSubDhi {
		panic("setPubSubParameters, must GossipSubDlo <= GossipSubD <= GossipSubDhi")
	}
	setOnce.Do(func() {
		pubsub.GossipSubDlo = psConf.GossipSubDlo
		pubsub.GossipSubD = psConf.GossipSubD
		pubsub.GossipSubDhi = psConf.GossipSubDhi
		pubsub.GossipSubHeartbeatInterval = time.Duration(psConf.GossipSubHeartbeatInterval) * time.Millisecond
		pubsub.GossipSubHistoryGossip = psConf.GossipSubHistoryGossip
		pubsub.GossipSubHistoryLength = psConf.GossipSubHistoryLength
	})
}
func NewPubSub(ctx context.Context, host host.Host, psConf *p2ptypes.PubSubConfig) (*PubSub, error) {
	p := &PubSub{
		topics: make(TopicMap),
	}
	setPubSubParameters(psConf)
	psOpts := make([]pubsub.Option, 0)
	if psConf.DisablePubSubMsgSign {
		psOpts = append(psOpts, pubsub.WithMessageSigning(false), pubsub.WithStrictSignatureVerification(false))
	}
	psOpts = append(psOpts, pubsub.WithFloodPublish(true),
		pubsub.WithPeerOutboundQueueSize(psConf.PeerOutboundQueueSize),
		pubsub.WithValidateQueueSize(psConf.ValidateQueueSize),
		pubsub.WithPeerExchange(psConf.EnablePeerExchange),
		pubsub.WithMaxMessageSize(psConf.MaxMsgSize))
	ps, err := pubsub.NewGossipSub(ctx, host, psOpts...)
	if err != nil {
		return nil, err
	}
	p.PubSub = ps
	p.ctx = ctx
	p.topics = make(TopicMap)
	return p, nil
}
func (p *PubSub) HasTopic(topic string) bool {
	p.topicMutex.RLock()
	defer p.topicMutex.RUnlock()
	_, ok := p.topics[topic]
	return ok
}
func (p *PubSub) JoinAndSubTopic(topic string, callback SubCallBack, opts ...pubsub.TopicOpt) error {
	Topic, err := p.Join(topic, opts...)
	if err != nil {
		return err
	}
	subscription, err := Topic.Subscribe()
	if err != nil {
		return err
	}
	ctx, cancel := context.WithCancel(p.ctx)
	p.topicMutex.Lock()
	p.topics[topic] = &topicinfo{
		pubtopic: Topic,
		ctx:      ctx,
		topic:    topic,
		cancel:   cancel,
		sub:      subscription,
	}
	p.topicMutex.Unlock()
	go p.subTopic(ctx, subscription, callback)
	return nil
}
func (p *PubSub) Publish(topic string, msg []byte) error {
	p.topicMutex.RLock()
	defer p.topicMutex.RUnlock()
	t, ok := p.topics[topic]
	if !ok {
		log.Error("pubsub publish", "no this topic", topic)
		return fmt.Errorf("no this topic:%v", topic)
	}
	err := t.pubtopic.Publish(t.ctx, msg)
	if err != nil {
		log.Error("pubsub publish", "err", err)
		return err
	}
	return nil
}
func (p *PubSub) subTopic(ctx context.Context, sub *pubsub.Subscription, callback SubCallBack) {
	topic := sub.Topic()
	for {
		got, err := sub.Next(ctx)
		if err != nil {
			log.Error("SubMsg", "topic", topic, "sub err", err)
			p.RemoveTopic(topic)
			return
		}
		callback(topic, got)
	}
}