pubsub.go
func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option) (*PubSub, error) {
	ps := &PubSub{
		host:             h,
		ctx:              ctx,
		rt:               rt,
		signID:           h.ID(),
		signKey:          h.Peerstore().PrivKey(h.ID()),
		signStrict:       true,
		incoming:         make(chan *RPC, 32),
		publish:          make(chan *Message),
		newPeers:         make(chan peer.ID),
		newPeerStream:    make(chan inet.Stream),
		newPeerError:     make(chan peer.ID),
		peerDead:         make(chan peer.ID),
		cancelCh:         make(chan *Subscription),
		getPeers:         make(chan *listPeerReq),
		addSub:           make(chan *addSubReq),
		getTopics:        make(chan *topicReq),
		sendMsg:          make(chan *sendReq, 32),
		addVal:           make(chan *addValReq),
		rmVal:            make(chan *rmValReq),
		validateThrottle: make(chan struct{}, defaultValidateThrottle),
		eval:             make(chan func()),
		myTopics:         make(map[string]map[*Subscription]struct{}),
		topics:           make(map[string]map[peer.ID]struct{}),
		peers:            make(map[peer.ID]chan *RPC),
		topicVals:        make(map[string]*topicVal),
		blacklist:        NewMapBlacklist(),
		blacklistPeer:    make(chan peer.ID),
		seenMessages:     timecache.NewTimeCache(TimeCacheDuration),
		counter:          uint64(time.Now().UnixNano()),
	}

	for _, opt := range opts {
		err := opt(ps)
		if err != nil {
			return nil, err
		}
	}

	if ps.signStrict && ps.signKey == nil {
		return nil, fmt.Errorf("strict signature verification enabled but message signing is disabled")
	}

	rt.Attach(ps)

	for _, id := range rt.Protocols() {
		h.SetStreamHandler(id, ps.handleNewStream)
	}
	h.Network().Notify((*PubSubNotif)(ps))

	go ps.processLoop(ctx)

	return ps, nil
}
func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option) (*PubSub, error) {
	ps := &PubSub{
		host:             h,
		ctx:              ctx,
		rt:               rt,
		signID:           h.ID(),
		signKey:          h.Peerstore().PrivKey(h.ID()),
		signStrict:       true,
		incoming:         make(chan *RPC, 32),
		publish:          make(chan *Message),
		newPeers:         make(chan peer.ID),
		newPeerStream:    make(chan inet.Stream),
		newPeerError:     make(chan peer.ID),
		peerDead:         make(chan peer.ID),
		cancelCh:         make(chan *Subscription),
		getPeers:         make(chan *listPeerReq),
		addSub:           make(chan *addSubReq),
		getTopics:        make(chan *topicReq),
		sendMsg:          make(chan *sendReq, 32),
		addVal:           make(chan *addValReq),
		rmVal:            make(chan *rmValReq),
		validateThrottle: make(chan struct{}, defaultValidateThrottle),
		eval:             make(chan func()),
		myTopics:         make(map[string]map[*Subscription]struct{}),
		topics:           make(map[string]map[peer.ID]struct{}),
		peers:            make(map[peer.ID]chan *RPC),
		topicVals:        make(map[string]*topicVal),
		blacklist:        NewMapBlacklist(),
		blacklistPeer:    make(chan peer.ID),
		seenMessages:     timecache.NewTimeCache(TimeCacheDuration),
		counter:          uint64(time.Now().UnixNano()),
	}

	for _, opt := range opts {
		err := opt(ps)
		if err != nil {
			return nil, err
		}
	}

	if ps.signStrict && ps.signKey == nil {
		return nil, fmt.Errorf("strict signature verification enabled but message signing is disabled")
	}

	rt.Attach(ps)

	for _, id := range rt.Protocols() {
		h.SetStreamHandler(id, ps.handleNewStream)
	}
	h.Network().Notify((*PubSubNotif)(ps))

	go ps.processLoop(ctx)

	return ps, nil
}
// NewPubSub returns a new PubSub management object.