/*
GetPeerSubscriptions is a API function which allows to query a peer for stream subscriptions it has.
It can be called via RPC.
It returns a map of node IDs with an array of string representations of Stream objects.
*/
556213
func (api *API) GetPeerSubscriptions() map[string][]string {
	//create the empty map
	pstreams := make(map[string][]string)

	//iterate all streamer peers
	api.streamer.peersMu.RLock()
	defer api.streamer.peersMu.RUnlock()

	for id, p := range api.streamer.peers {
		var streams []string
		//every peer has a map of stream servers
		//every stream server represents a subscription
		p.serverMu.RLock()
		for s := range p.servers {
			//append the string representation of the stream
			//to the list for this peer
			streams = append(streams, s.String())
		}
		p.serverMu.RUnlock()
		//set the array of stream servers to the map
		pstreams[id.String()] = streams
	}
	return pstreams
}
49191
func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8) error {
	// check if the stream is registered
	if _, err := r.GetClientFunc(s.Name); err != nil {
		return err
	}

	peer := r.getPeer(peerId)
	if peer == nil {
		return fmt.Errorf("peer not found %v", peerId)
	}

	var to uint64
	if !s.Live && h != nil {
		to = h.To
	}

	err := peer.setClientParams(s, newClientParams(priority, to))
	if err != nil {
		return err
	}
	if s.Live && h != nil {
		if err := peer.setClientParams(
			getHistoryStream(s),
			newClientParams(getHistoryPriority(priority), h.To),
		); err != nil {
			return err
		}
	}

	msg := &SubscribeMsg{
		Stream:   s,
		History:  h,
		Priority: priority,
	}
	log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)

	return peer.SendPriority(context.TODO(), msg, priority)
}
1104032
func newSubscription(provider *SubscriptionProvider) *Subscription {
	return &Subscription{
		id:       uuid.NewV4(),
		events:   make(chan interface{}, subscriptionChanSize),
		quitChan: make(chan struct{}),
		provider: provider,
	}
}
852269
func (gc *gossipChannel) GetPeers() []discovery.NetworkMember {
	var members []discovery.NetworkMember
	if gc.hasLeftChannel() {
		return members
	}

	for _, member := range gc.GetMembership() {
		if !gc.EligibleForChannel(member) {
			continue
		}
		stateInf := gc.stateInfoMsgStore.MsgByID(member.PKIid)
		if stateInf == nil {
			continue
		}
		props := stateInf.GetStateInfo().Properties
		if props != nil && props.LeftChannel {
			continue
		}
		member.Properties = stateInf.GetStateInfo().Properties
		member.Envelope = stateInf.Envelope
		members = append(members, member)
	}
	return members
}
18298
func getChannels() pb.Response {
	channelInfoArray := peer.GetChannelsInfo()

	// add array with info about all channels for this peer
	cqr := &pb.ChannelQueryResponse{Channels: channelInfoArray}

	cqrbytes, err := proto.Marshal(cqr)
	if err != nil {
		return shim.Error(err.Error())
	}

	return shim.Success(cqrbytes)
}
1146297
func getEventSubscriptionHandler(manager *golongpoll.LongpollManager) func(w http.ResponseWriter, r *http.Request) {
	// Creates closure that captures the LongpollManager
	// Wraps the manager.SubscriptionHandler with a layer of dummy access control validation
	return func(w http.ResponseWriter, r *http.Request) {
		category := r.URL.Query().Get("category")
		user := r.URL.Query().Get("user")
		// NOTE: real user authentication should be used in the real world!

		// Dummy user access control in the event the client is requesting
		// a user's private activity stream:
		if category == "larry_actions" && user != "larry" {
			w.WriteHeader(http.StatusForbidden)
			w.Write([]byte("You're not Larry."))
			return
		}
		if category == "moe_actions" && user != "moe" {
			w.WriteHeader(http.StatusForbidden)
			w.Write([]byte("You're not Moe."))
			return
		}
		if category == "curly_actions" && user != "curly" {
			w.WriteHeader(http.StatusForbidden)
			w.Write([]byte("You're not Curly."))
			return
		}

		// Only allow supported subscription categories:
		if category != "public_actions" && category != "larry_actions" &&
			category != "moe_actions" && category != "curly_actions" {
			w.WriteHeader(http.StatusBadRequest)
			w.Write([]byte("Subscription channel does not exist."))
			return
		}

		// Client is either requesting the public stream, or a private
		// stream that they're allowed to see.
		// Go ahead and let the subscription happen:
		manager.SubscriptionHandler(w, r)
	}
}
1206731
func (r *Repository) GetSubscriptionURL() string {
	if r == nil || r.SubscriptionURL == nil {
		return ""
	}
	return *r.SubscriptionURL
}
720918
func (h *handler) handleSubscriptionResult(msg *jsonrpcMessage) {
	var result subscriptionResult
	if err := json.Unmarshal(msg.Params, &result); err != nil {
		h.log.Debug("Dropping invalid subscription message")
		return
	}
	if h.clientSubs[result.ID] != nil {
		h.clientSubs[result.ID].deliver(result.Result)
	}
}
207255
func countPsubs(subs []*Subscriber) int {
	n := 0
	for _, p := range subs {
		n += len(p.patterns)
	}
	return n
}
686154
public List<Subscription> getSubscriptions() throws NoResponseException, XMPPErrorException, NotConnectedException, InterruptedException {
        Stanza reply = sendPubsubPacket(Type.get, new NodeExtension(PubSubElementType.SUBSCRIPTIONS), null);
        SubscriptionsExtension subElem = reply.getExtension(PubSubElementType.SUBSCRIPTIONS.getElementName(), PubSubElementType.SUBSCRIPTIONS.getNamespace().getXmlns());
        return subElem.getSubscriptions();
    }
1108577
func (m *Miniredis) PubSubChannels(pattern string) []string {
	m.Lock()
	defer m.Unlock()

	return activeChannels(m.allSubscribers(), pattern)
}
1117674
func (e *Exporter) getPeers() error {
	peers := e.service.Peers()
	e.datum.TotalPeers = float64(len(peers))
	e.datum.InboundPeers = 0
	e.datum.OutboundPeers = 0

	for _, peer := range peers {
		if peer.IsOutbound {
			e.datum.OutboundPeers += 1
		} else {
			e.datum.InboundPeers += 1
		}
	}

	return nil
}
152387
func (r *Redis) PubSub() (*PubSub, error) {
	c, err := r.pool.Get()
	if err != nil {
		return nil, err
	}
	return &PubSub{
		redis:    r,
		conn:     c,
		Patterns: make(map[string]bool),
		Channels: make(map[string]bool),
	}, nil
}
82409
func (c *Client) GetAllPublicEndpoints() ([]service.PublicEndpoint, error) {
	var response []service.PublicEndpoint
	if err := c.call("GetAllPublicEndpoints", empty, &response); err != nil {
		return response, err
	}
	return response, nil
}
20561
func (c *Client) LookupSubscription(subID string) (*SubscriptionRequest, error) {
	s, err := c.subscriptions.lookupBySubscriptionID(subID)
	if err != nil {
		return nil, err
	}
	return s.Request, nil
}
903041
func (pm *peersManager) GetPeers(infoHash string, size int) []*Peer {
	peers := make([]*Peer, 0, size)

	v, ok := pm.table.Get(infoHash)
	if !ok {
		return peers
	}

	for e := range v.(*keyedDeque).Iter() {
		peers = append(peers, e.Value.(*Peer))
	}

	if len(peers) > size {
		peers = peers[len(peers)-size:]
	}
	return peers
}
820202
func (ds *MockStaticDiscoveryService) GetPeers() ([]fab.Peer, error) {

	if ds.Error != nil {
		return nil, ds.Error
	}

	if ds.Peers == nil {
		mockPeer := MockPeer{MockName: "Peer1", MockURL: "http://peer1.com", MockRoles: []string{}, MockCert: nil, MockMSP: "Org1MSP"}
		peers := make([]fab.Peer, 0)
		peers = append(peers, &mockPeer)
		ds.Peers = peers
	}

	return ds.Peers, nil

}