// SubscribeNewPeers subscribes to peer arrival events.
328221
func (ps *peerSet) SubscribeNewPeers(ch chan<- *peerConnection) event.Subscription {
	return ps.newPeerFeed.Subscribe(ch)
}
1337087
func (r *rpcServer) SubscribeChannelEvents(req *lnrpc.ChannelEventSubscription,
	updateStream lnrpc.Lightning_SubscribeChannelEventsServer) error {

	channelEventSub, err := r.server.channelNotifier.SubscribeChannelEvents()
	if err != nil {
		return err
	}

	// Ensure that the resources for the client is cleaned up once either
	// the server, or client exits.
	defer channelEventSub.Cancel()

	graph := r.server.chanDB.ChannelGraph()

	for {
		select {
		// A new update has been sent by the channel router, we'll
		// marshal it into the form expected by the gRPC client, then
		// send it off to the client(s).
		case e := <-channelEventSub.Updates():
			var update *lnrpc.ChannelEventUpdate
			switch event := e.(type) {
			case channelnotifier.OpenChannelEvent:
				channel := createRPCOpenChannel(r, graph,
					event.Channel, true)
				update = &lnrpc.ChannelEventUpdate{
					Type: lnrpc.ChannelEventUpdate_OPEN_CHANNEL,
					Channel: &lnrpc.ChannelEventUpdate_OpenChannel{
						OpenChannel: channel,
					},
				}
			case channelnotifier.ClosedChannelEvent:
				closedChannel := createRPCClosedChannel(event.CloseSummary)
				update = &lnrpc.ChannelEventUpdate{
					Type: lnrpc.ChannelEventUpdate_CLOSED_CHANNEL,
					Channel: &lnrpc.ChannelEventUpdate_ClosedChannel{
						ClosedChannel: closedChannel,
					},
				}
			case channelnotifier.ActiveChannelEvent:
				update = &lnrpc.ChannelEventUpdate{
					Type: lnrpc.ChannelEventUpdate_ACTIVE_CHANNEL,
					Channel: &lnrpc.ChannelEventUpdate_ActiveChannel{
						ActiveChannel: &lnrpc.ChannelPoint{
							FundingTxid: &lnrpc.ChannelPoint_FundingTxidBytes{
								FundingTxidBytes: event.ChannelPoint.Hash[:],
							},
							OutputIndex: event.ChannelPoint.Index,
						},
					},
				}
			case channelnotifier.InactiveChannelEvent:
				update = &lnrpc.ChannelEventUpdate{
					Type: lnrpc.ChannelEventUpdate_INACTIVE_CHANNEL,
					Channel: &lnrpc.ChannelEventUpdate_InactiveChannel{
						InactiveChannel: &lnrpc.ChannelPoint{
							FundingTxid: &lnrpc.ChannelPoint_FundingTxidBytes{
								FundingTxidBytes: event.ChannelPoint.Hash[:],
							},
							OutputIndex: event.ChannelPoint.Index,
						},
					},
				}
			default:
				return fmt.Errorf("unexpected channel event update: %v", event)
			}

			if err := updateStream.Send(update); err != nil {
				return err
			}
		case <-r.quit:
			return nil
		}
	}
}
733587
func (s *CreateBGPPeerInput) SetNewBGPPeer(v *NewBGPPeer) *CreateBGPPeerInput {
	s.NewBGPPeer = v
	return s
}
1335315
func (s *server) handlePeers(w http.ResponseWriter, r *http.Request) {
	peers := make([]*protocol.Peer, 0, len(s.network.Peers))
	for _, peer := range s.network.Peers {
		peers = append(peers, peer.Peer)
	}
	json.NewEncoder(w).Encode(peers)
}
574899
func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
	p := &Peer{
		Peer:         peer,
		pq:           pq.New(int(PriorityQueue), PriorityQueueCap),
		streamer:     streamer,
		servers:      make(map[Stream]*server),
		clients:      make(map[Stream]*client),
		clientParams: make(map[Stream]*clientParams),
		quit:         make(chan struct{}),
	}
	ctx, cancel := context.WithCancel(context.Background())
	go p.pq.Run(ctx, func(i interface{}) {
		wmsg := i.(WrappedPriorityMsg)
		err := p.Send(wmsg.Context, wmsg.Msg)
		if err != nil {
			log.Error("Message send error, dropping peer", "peer", p.ID(), "err", err)
			p.Drop(err)
		}
	})

	// basic monitoring for pq contention
	go func(pq *pq.PriorityQueue) {
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				var lenMaxi int
				var capMaxi int
				for k := range pq.Queues {
					if lenMaxi < len(pq.Queues[k]) {
						lenMaxi = len(pq.Queues[k])
					}

					if capMaxi < cap(pq.Queues[k]) {
						capMaxi = cap(pq.Queues[k])
					}
				}

				metrics.GetOrRegisterGauge(fmt.Sprintf("pq_len_%s", p.ID().TerminalString()), nil).Update(int64(lenMaxi))
				metrics.GetOrRegisterGauge(fmt.Sprintf("pq_cap_%s", p.ID().TerminalString()), nil).Update(int64(capMaxi))
			case <-p.quit:
				return
			}
		}
	}(p.pq)

	go func() {
		<-p.quit

		cancel()
	}()
	return p
}
624467
func (s *DiscoveryWrapper) GetPeers() ([]fab.Peer, error) {
	var eventEndpoints []fab.Peer

	peers, err := s.DiscoveryService.GetPeers()
	if err != nil {
		return nil, err
	}

	for _, peer := range peers {

		var peerConfig *fab.PeerConfig
		var found bool

		chPeer := s.getChannelPeer(peer.URL())
		if chPeer != nil {
			peerConfig = &chPeer.PeerConfig
		} else {
			peerConfig, found = s.ctx.EndpointConfig().PeerConfig(peer.URL())
			if !found {
				continue
			}
			chPeer = s.getChannelPeer(peerConfig.URL)
		}

		logger.Debugf("Channel peer config for [%s]: %#v", peer.URL(), chPeer)

		if chPeer != nil && !chPeer.EventSource {
			logger.Debugf("Excluding peer [%s] since it is not configured as an event source", peer.URL())
			continue
		}

		eventEndpoint := FromPeerConfig(s.ctx.EndpointConfig(), peer, peerConfig)
		eventEndpoints = append(eventEndpoints, eventEndpoint)
	}

	return eventEndpoints, nil
}
93413
func (p *Provider) Subscribe(opts pubsub.HandlerOptions, h pubsub.MsgHandler) {
	logrus.Infof("Subscribing to %s successWriter/name %s", opts.Topic, opts.ServiceName+"."+opts.Name)
	logSuccess := logrus.WithField("pubsub", "kafka")
	successWriter := logSuccess.Writer()
	p.Reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        p.Brokers,
		GroupID:        opts.ServiceName + "." + opts.Name,
		Topic:          opts.Topic,
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
		MaxWait:        50 * time.Millisecond,
		Logger:         log.New(successWriter, "", log.Lshortfile),
		ErrorLogger:    log.New(ioutil.Discard, "", log.Lshortfile),
	})

	b := &backoff.Backoff{
		Min:    200 * time.Millisecond,
		Max:    600 * time.Second,
		Factor: 2,
		Jitter: true,
	}

	go func() {
		for {
			ctx := context.Background()
			m, err := p.Reader.FetchMessage(ctx)
			if err != nil {
				d := b.Duration()
				logrus.Errorf(
					"Subscription receive to topic %s failed, reconnecting in %v. Err: %v",
					opts.Topic, d, err,
				)
				time.Sleep(d)
			}

			b.Reset()

			msg := pubsub.Msg{
				ID:   string(m.Key),
				Data: m.Value,
				Ack: func() {
					p.Reader.CommitMessages(ctx, m)
				},
				Nack: func() {},
			}

			err = h(ctx, msg)
			if err != nil {
				break
			}

			if opts.AutoAck {
				msg.Ack()
			}

			logrus.Debugf("message at topic/partition/offset %v/%v/%v\n",
				m.Topic, m.Partition, m.Offset)
		}
		successWriter.Close()
	}()
}
749786
func (pn *peernet) handleNewStream(s inet.Stream) {
	pn.RLock()
	handler := pn.streamHandler
	pn.RUnlock()
	if handler != nil {
		go handler(s)
	}
}
661010
func (sm *SyncManager) NewPeer(peer *peerpkg.Peer) {
	// Ignore if we are shutting down.
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}
	sm.msgChan <- &newPeerMsg{peer: peer}
}
304994
func (p *PubSub) PSubscribe(patterns ...string) error {
	args := packArgs("PSUBSCRIBE", patterns)
	return p.conn.SendCommand(args...)
}
727571
func (t *Transport) onPeerStatusChanged(tp *tchannel.Peer) {
	t.lock.Lock()
	defer t.lock.Unlock()

	p, ok := t.peers[tp.HostPort()]
	if !ok {
		return
	}
	p.OnStatusChanged()
}
115055
func (ec *EthereumClient) SubscribeNewHead(ctx *Context, handler NewHeadHandler, buffer int) (sub *Subscription, _ error) {
	// Subscribe to the event internally
	ch := make(chan *types.Header, buffer)
	rawSub, err := ec.client.SubscribeNewHead(ctx.context, ch)
	if err != nil {
		return nil, err
	}
	// Start up a dispatcher to feed into the callback
	go func() {
		for {
			select {
			case header := <-ch:
				handler.OnNewHead(&Header{header})

			case err := <-rawSub.Err():
				if err != nil {
					handler.OnError(err.Error())
				}
				return
			}
		}
	}()
	return &Subscription{rawSub}, nil
}
294665
func newSubscriber() *Subscriber {
	return &Subscriber{
		publish:  make(chan PubsubMessage),
		channels: map[string]struct{}{},
		patterns: map[string]*regexp.Regexp{},
	}
}
797775
func (ps *peerSet) SubscribePeerDrops(ch chan<- *peerConnection) event.Subscription {
	return ps.peerDropFeed.Subscribe(ch)
}
718356
func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool) (*Subscription, error) {
	if nc == nil {
		return nil, ErrInvalidConnection
	}
	nc.mu.Lock()
	// ok here, but defer is generally expensive
	defer nc.mu.Unlock()

	// Check for some error conditions.
	if nc.isClosed() {
		return nil, ErrConnectionClosed
	}
	if nc.isDraining() {
		return nil, ErrConnectionDraining
	}

	if cb == nil && ch == nil {
		return nil, ErrBadSubscription
	}

	sub := &Subscription{Subject: subj, Queue: queue, mcb: cb, conn: nc}
	// Set pending limits.
	sub.pMsgsLimit = DefaultSubPendingMsgsLimit
	sub.pBytesLimit = DefaultSubPendingBytesLimit

	// If we have an async callback, start up a sub specific
	// Go routine to deliver the messages.
	if cb != nil {
		sub.typ = AsyncSubscription
		sub.pCond = sync.NewCond(&sub.mu)
		go nc.waitForMsgs(sub)
	} else if !isSync {
		sub.typ = ChanSubscription
		sub.mch = ch
	} else { // Sync Subscription
		sub.typ = SyncSubscription
		sub.mch = ch
	}

	nc.subsMu.Lock()
	nc.ssid++
	sub.sid = nc.ssid
	nc.subs[sub.sid] = sub
	nc.subsMu.Unlock()

	// We will send these for all subs when we reconnect
	// so that we can suppress here if reconnecting.
	if !nc.isReconnecting() {
		fmt.Fprintf(nc.bw, subProto, subj, queue, sub.sid)
		// Kick flusher if needed.
		if len(nc.fch) == 0 {
			nc.kickFlusher()
		}
	}

	return sub, nil
}
1152815
func (sn *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
	srv := sn.Server()
	if srv == nil {
		panic("node not running")
	}
	return srv.SubscribeEvents(ch)
}
913599
func NewPeerServer(listenAddress string, serverConfig comm.ServerConfig) (*comm.GRPCServer, error) {
	var err error
	peerServer, err = comm.NewGRPCServer(listenAddress, serverConfig)
	if err != nil {
		peerLogger.Errorf("Failed to create peer server (%s)", err)
		return nil, err
	}
	return peerServer, nil
}