func (p *PubSub) processLoop(ctx context.Context) {
	defer func() {
		// Clean up go routines.
		for _, ch := range p.peers {
			close(ch)
		}
		p.peers = nil
		p.topics = nil
	}()

	for {
		select {
		case pid := <-p.newPeers:
			if _, ok := p.peers[pid]; ok {
				log.Warning("already have connection to peer: ", pid)
				continue
			}

			if p.blacklist.Contains(pid) {
				log.Warning("ignoring connection from blacklisted peer: ", pid)
				continue
			}

			messages := make(chan *RPC, 32)
			messages <- p.getHelloPacket()
			go p.handleNewPeer(ctx, pid, messages)
			p.peers[pid] = messages

		case s := <-p.newPeerStream:
			pid := s.Conn().RemotePeer()

			ch, ok := p.peers[pid]
			if !ok {
				log.Warning("new stream for unknown peer: ", pid)
				s.Reset()
				continue
			}

			if p.blacklist.Contains(pid) {
				log.Warning("closing stream for blacklisted peer: ", pid)
				close(ch)
				s.Reset()
				continue
			}

			p.rt.AddPeer(pid, s.Protocol())

		case pid := <-p.newPeerError:
			delete(p.peers, pid)

		case pid := <-p.peerDead:
			ch, ok := p.peers[pid]
			if !ok {
				continue
			}

			close(ch)

			if p.host.Network().Connectedness(pid) == inet.Connected {
				// still connected, must be a duplicate connection being closed.
				// we respawn the writer as we need to ensure there is a stream active
				log.Warning("peer declared dead but still connected; respawning writer: ", pid)
				messages := make(chan *RPC, 32)
				messages <- p.getHelloPacket()
				go p.handleNewPeer(ctx, pid, messages)
				p.peers[pid] = messages
				continue
			}

			delete(p.peers, pid)
			for _, t := range p.topics {
				delete(t, pid)
			}

			p.rt.RemovePeer(pid)

		case treq := <-p.getTopics:
			var out []string
			for t := range p.myTopics {
				out = append(out, t)
			}
			treq.resp <- out
		case sub := <-p.cancelCh:
			p.handleRemoveSubscription(sub)
		case sub := <-p.addSub:
			p.handleAddSubscription(sub)
		case preq := <-p.getPeers:
			tmap, ok := p.topics[preq.topic]
			if preq.topic != "" && !ok {
				preq.resp <- nil
				continue
			}
			var peers []peer.ID
			for p := range p.peers {
				if preq.topic != "" {
					_, ok := tmap[p]
					if !ok {
						continue
					}
				}
				peers = append(peers, p)
			}
			preq.resp <- peers
		case rpc := <-p.incoming:
			p.handleIncomingRPC(rpc)

		case msg := <-p.publish:
			vals := p.getValidators(msg)
			p.pushMsg(vals, p.host.ID(), msg)

		case req := <-p.sendMsg:
			p.publishMessage(req.from, req.msg.Message)

		case req := <-p.addVal:
			p.addValidator(req)

		case req := <-p.rmVal:
			p.rmValidator(req)

		case thunk := <-p.eval:
			thunk()

		case pid := <-p.blacklistPeer:
			log.Infof("Blacklisting peer %s", pid)
			p.blacklist.Add(pid)

			ch, ok := p.peers[pid]
			if ok {
				close(ch)
				delete(p.peers, pid)
				for _, t := range p.topics {
					delete(t, pid)
				}
				p.rt.RemovePeer(pid)
			}

		case <-ctx.Done():
			log.Info("pubsub processloop shutting down")
			return
		}
	}
}
func (p *PubSub) processLoop(ctx context.Context) {
	defer func() {
		
		for _, ch := range p.peers {
			close(ch)
		}
		p.peers = nil
		p.topics = nil
	}()

	for {
		select {
		case pid := <-p.newPeers:
			if _, ok := p.peers[pid]; ok {
				log.Warning("already have connection to peer: ", pid)
				continue
			}

			if p.blacklist.Contains(pid) {
				log.Warning("ignoring connection from blacklisted peer: ", pid)
				continue
			}

			messages := make(chan *RPC, 32)
			messages <- p.getHelloPacket()
			go p.handleNewPeer(ctx, pid, messages)
			p.peers[pid] = messages

		case s := <-p.newPeerStream:
			pid := s.Conn().RemotePeer()

			ch, ok := p.peers[pid]
			if !ok {
				log.Warning("new stream for unknown peer: ", pid)
				s.Reset()
				continue
			}

			if p.blacklist.Contains(pid) {
				log.Warning("closing stream for blacklisted peer: ", pid)
				close(ch)
				s.Reset()
				continue
			}

			p.rt.AddPeer(pid, s.Protocol())

		case pid := <-p.newPeerError:
			delete(p.peers, pid)

		case pid := <-p.peerDead:
			ch, ok := p.peers[pid]
			if !ok {
				continue
			}

			close(ch)

			if p.host.Network().Connectedness(pid) == inet.Connected {
				
				
				log.Warning("peer declared dead but still connected; respawning writer: ", pid)
				messages := make(chan *RPC, 32)
				messages <- p.getHelloPacket()
				go p.handleNewPeer(ctx, pid, messages)
				p.peers[pid] = messages
				continue
			}

			delete(p.peers, pid)
			for _, t := range p.topics {
				delete(t, pid)
			}

			p.rt.RemovePeer(pid)

		case treq := <-p.getTopics:
			var out []string
			for t := range p.myTopics {
				out = append(out, t)
			}
			treq.resp <- out
		case sub := <-p.cancelCh:
			p.handleRemoveSubscription(sub)
		case sub := <-p.addSub:
			p.handleAddSubscription(sub)
		case preq := <-p.getPeers:
			tmap, ok := p.topics[preq.topic]
			if preq.topic != "" && !ok {
				preq.resp <- nil
				continue
			}
			var peers []peer.ID
			for p := range p.peers {
				if preq.topic != "" {
					_, ok := tmap[p]
					if !ok {
						continue
					}
				}
				peers = append(peers, p)
			}
			preq.resp <- peers
		case rpc := <-p.incoming:
			p.handleIncomingRPC(rpc)

		case msg := <-p.publish:
			vals := p.getValidators(msg)
			p.pushMsg(vals, p.host.ID(), msg)

		case req := <-p.sendMsg:
			p.publishMessage(req.from, req.msg.Message)

		case req := <-p.addVal:
			p.addValidator(req)

		case req := <-p.rmVal:
			p.rmValidator(req)

		case thunk := <-p.eval:
			thunk()

		case pid := <-p.blacklistPeer:
			log.Infof("Blacklisting peer %s", pid)
			p.blacklist.Add(pid)

			ch, ok := p.peers[pid]
			if ok {
				close(ch)
				delete(p.peers, pid)
				for _, t := range p.topics {
					delete(t, pid)
				}
				p.rt.RemovePeer(pid)
			}

		case <-ctx.Done():
			log.Info("pubsub processloop shutting down")
			return
		}
	}
}
func ( p * PubSub ) processLoop ( ctx context . Context ) { defer func ( ) { // Clean up go routines. for _ , ch := range p . peers { close ( ch ) } p . peers = nil p . topics = nil } ( ) for { select { case pid := <- p . newPeers : if _ , ok := p . peers [ pid ] ; ok { log . Warning ( " " , pid ) continue } if p . blacklist . Contains ( pid ) { log . Warning ( " " , pid ) continue } messages := make ( chan * RPC , 32 ) messages <- p . getHelloPacket ( ) go p . handleNewPeer ( ctx , pid , messages ) p . peers [ pid ] = messages case s := <- p . newPeerStream : pid := s . Conn ( ) . RemotePeer ( ) ch , ok := p . peers [ pid ] if ! ok { log . Warning ( " " , pid ) s . Reset ( ) continue } if p . blacklist . Contains ( pid ) { log . Warning ( " " , pid ) close ( ch ) s . Reset ( ) continue } p . rt . AddPeer ( pid , s . Protocol ( ) ) case pid := <- p . newPeerError : delete ( p . peers , pid ) case pid := <- p . peerDead : ch , ok := p . peers [ pid ] if ! ok { continue } close ( ch ) if p . host . Network ( ) . Connectedness ( pid ) == inet . Connected { // still connected, must be a duplicate connection being closed. // we respawn the writer as we need to ensure there is a stream active log . Warning ( " " , pid ) messages := make ( chan * RPC , 32 ) messages <- p . getHelloPacket ( ) go p . handleNewPeer ( ctx , pid , messages ) p . peers [ pid ] = messages continue } delete ( p . peers , pid ) for _ , t := range p . topics { delete ( t , pid ) } p . rt . RemovePeer ( pid ) case treq := <- p . getTopics : var out [ ] string for t := range p . myTopics { out = append ( out , t ) } treq . resp <- out case sub := <- p . cancelCh : p . handleRemoveSubscription ( sub ) case sub := <- p . addSub : p . handleAddSubscription ( sub ) case preq := <- p . getPeers : tmap , ok := p . topics [ preq . topic ] if preq . topic != " " && ! ok { preq . resp <- nil continue } var peers [ ] peer . ID for p := range p . peers { if preq . topic != " " { _ , ok := tmap [ p ] if ! ok { continue } } peers = append ( peers , p ) } preq . resp <- peers case rpc := <- p . incoming : p . handleIncomingRPC ( rpc ) case msg := <- p . publish : vals := p . getValidators ( msg ) p . pushMsg ( vals , p . host . ID ( ) , msg ) case req := <- p . sendMsg : p . publishMessage ( req . from , req . msg . Message ) case req := <- p . addVal : p . addValidator ( req ) case req := <- p . rmVal : p . rmValidator ( req ) case thunk := <- p . eval : thunk ( ) case pid := <- p . blacklistPeer : log . Infof ( " " , pid ) p . blacklist . Add ( pid ) ch , ok := p . peers [ pid ] if ok { close ( ch ) delete ( p . peers , pid ) for _ , t := range p . topics { delete ( t , pid ) } p . rt . RemovePeer ( pid ) } case <- ctx . Done ( ) : log . Info ( " " ) return } } }
// processLoop handles all inputs arriving on the channels
processLoop handles all inputs arriving on the channels