// Return all local client subscriptions. Use the supplied slice.
348888
func (s *Sublist) localSubs(subs *[]*subscription) {
	s.RLock()
	s.collectLocalSubs(s.root, subs)
	s.RUnlock()
}
47866
func (s *Sublist) All(subs *[]*subscription) {
	s.RLock()
	s.collectAllSubs(s.root, subs)
	s.RUnlock()
}
371820
func (self *Client) Subscribe(channel string) (ch *Channel) {
	for _, ch := range self.Channels {
		if ch.Name == channel {
			self._subscribe <- ch
			return ch
		}
	}
	ch = &Channel{Name: channel, bindings: &self.bindings}
	self._subscribe <- ch
	return
}
10149
func (s *userService) GetAll() []datamodels.User {
	return s.repo.SelectMany(func(_ datamodels.User) bool {
		return true
	}, -1)
}
880897
func (n *Node) GetClientAllocs(args *structs.NodeSpecificRequest,
	reply *structs.NodeClientAllocsResponse) error {
	if done, err := n.srv.forward("Node.GetClientAllocs", args, args, reply); done {
		// We have a valid node connection since there is no error from the
		// forwarded server, so add the mapping to cache the
		// connection and allow the server to send RPCs to the client.
		if err == nil && n.ctx != nil && n.ctx.NodeID == "" {
			n.ctx.NodeID = args.NodeID
			n.srv.addNodeConn(n.ctx)
		}

		return err
	}
	defer metrics.MeasureSince([]string{"nomad", "client", "get_client_allocs"}, time.Now())

	// Verify the arguments
	if args.NodeID == "" {
		return fmt.Errorf("missing node ID")
	}

	// numOldAllocs is used to detect if there is a garbage collection event
	// that effects the node. When an allocation is garbage collected, that does
	// not change the modify index changes and thus the query won't unblock,
	// even though the set of allocations on the node has changed.
	var numOldAllocs int

	// Setup the blocking query
	opts := blockingOptions{
		queryOpts: &args.QueryOptions,
		queryMeta: &reply.QueryMeta,
		run: func(ws memdb.WatchSet, state *state.StateStore) error {
			// Look for the node
			node, err := state.NodeByID(ws, args.NodeID)
			if err != nil {
				return err
			}

			var allocs []*structs.Allocation
			if node != nil {
				if args.SecretID == "" {
					return fmt.Errorf("missing node secret ID for client status update")
				} else if args.SecretID != node.SecretID {
					return fmt.Errorf("node secret ID does not match")
				}

				// We have a valid node connection, so add the mapping to cache the
				// connection and allow the server to send RPCs to the client. We only cache
				// the connection if it is not being forwarded from another server.
				if n.ctx != nil && n.ctx.NodeID == "" && !args.IsForwarded() {
					n.ctx.NodeID = args.NodeID
					n.srv.addNodeConn(n.ctx)
				}

				var err error
				allocs, err = state.AllocsByNode(ws, args.NodeID)
				if err != nil {
					return err
				}
			}

			reply.Allocs = make(map[string]uint64)
			reply.MigrateTokens = make(map[string]string)

			// preferTableIndex is used to determine whether we should build the
			// response index based on the full table indexes versus the modify
			// indexes of the allocations on the specific node. This is
			// preferred in the case that the node doesn't yet have allocations
			// or when we detect a GC that effects the node.
			preferTableIndex := true

			// Setup the output
			if numAllocs := len(allocs); numAllocs != 0 {
				preferTableIndex = false

				for _, alloc := range allocs {
					reply.Allocs[alloc.ID] = alloc.AllocModifyIndex

					// If the allocation is going to do a migration, create a
					// migration token so that the client can authenticate with
					// the node hosting the previous allocation.
					if alloc.ShouldMigrate() {
						prevAllocation, err := state.AllocByID(ws, alloc.PreviousAllocation)
						if err != nil {
							return err
						}

						if prevAllocation != nil && prevAllocation.NodeID != alloc.NodeID {
							allocNode, err := state.NodeByID(ws, prevAllocation.NodeID)
							if err != nil {
								return err
							}
							if allocNode == nil {
								// Node must have been GC'd so skip the token
								continue
							}

							token, err := structs.GenerateMigrateToken(prevAllocation.ID, allocNode.SecretID)
							if err != nil {
								return err
							}
							reply.MigrateTokens[alloc.ID] = token
						}
					}

					reply.Index = maxUint64(reply.Index, alloc.ModifyIndex)
				}

				// Determine if we have less allocations than before. This
				// indicates there was a garbage collection
				if numAllocs < numOldAllocs {
					preferTableIndex = true
				}

				// Store the new number of allocations
				numOldAllocs = numAllocs
			}

			if preferTableIndex {
				// Use the last index that affected the nodes table
				index, err := state.Index("allocs")
				if err != nil {
					return err
				}

				// Must provide non-zero index to prevent blocking
				// Index 1 is impossible anyways (due to Raft internals)
				if index == 0 {
					reply.Index = 1
				} else {
					reply.Index = index
				}
			}
			return nil
		}}
	return n.srv.blockingRPC(&opts)
}
1308170
func (e *EmailsService) All() (emails []Email, result *Result) {
	result = e.client.get(e.URL, &emails)
	return
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
1144381
func (_class PGPUClass) GetAll(sessionID SessionRef) (_retval []PGPURef, _err error) {
	_method := "PGPU.get_all"
	_sessionIDArg, _err := convertSessionRefToXen(fmt.Sprintf("%s(%s)", _method, "session_id"), sessionID)
	if _err != nil {
		return
	}
	_result, _err := _class.client.APICall(_method, _sessionIDArg)
	if _err != nil {
		return
	}
	_retval, _err = convertPGPURefSetToGo(_method + " -> ", _result.Value)
	return
}
739009
func (_m *DBClient) GetAllProvisionWatchers() ([]models.ProvisionWatcher, error) {
	ret := _m.Called()

	var r0 []models.ProvisionWatcher
	if rf, ok := ret.Get(0).(func() []models.ProvisionWatcher); ok {
		r0 = rf()
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).([]models.ProvisionWatcher)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func() error); ok {
		r1 = rf()
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}
1210694
func (s *WebsocketServer) GetLocalClient(realm string, details map[string]interface{}) (*Client, error) {
	peer, err := s.Router.GetLocalPeer(URI(realm), details)
	if err != nil {
		return nil, err
	}
	c := NewClient(peer)
	go c.Receive()
	return c, nil
}
490494
func (s *IPServiceList) GetAllServices() ([]IPService, error) {
	services := s.Services
	t := s

	for t.HasNextPage() {
		var err error
		t, err = t.NextPage()
		if err != nil {
			return nil, err
		}
		services = append(services, t.Services...)
	}
	return services, nil
}
1004635
func (_class NetworkSriovClass) GetAll(sessionID SessionRef) (_retval []NetworkSriovRef, _err error) {
	_method := "network_sriov.get_all"
	_sessionIDArg, _err := convertSessionRefToXen(fmt.Sprintf("%s(%s)", _method, "session_id"), sessionID)
	if _err != nil {
		return
	}
	_result, _err := _class.client.APICall(_method, _sessionIDArg)
	if _err != nil {
		return
	}
	_retval, _err = convertNetworkSriovRefSetToGo(_method + " -> ", _result.Value)
	return
}
1013847
func (c *Local) resubscribe(subscriber string, q tmpubsub.Query) types.Subscription {
	attempts := 0
	for {
		if !c.IsRunning() {
			return nil
		}

		sub, err := c.EventBus.Subscribe(context.Background(), subscriber, q)
		if err == nil {
			return sub
		}

		attempts++
		time.Sleep((10 << uint(attempts)) * time.Millisecond) // 10ms -> 20ms -> 40ms
	}
}
172515
func (c *Client) GetRegions() ([]Region, error) {
	var regionMap map[string]Region
	if err := c.get(`regions/list`, &regionMap); err != nil {
		return nil, err
	}

	var regionList []Region
	for _, os := range regionMap {
		regionList = append(regionList, os)
	}
	sort.Sort(regions(regionList))
	return regionList, nil
}
52918
func (p *PullRequestsService) All() (pulls []PullRequest, result *Result) {
	result = p.client.get(p.URL, &pulls)
	return
}
144095
func (i *InvoiceRegistry) HodlUnsubscribeAll(subscriber chan<- interface{}) {
	i.Lock()
	defer i.Unlock()

	hashes := i.hodlReverseSubscriptions[subscriber]
	for hash := range hashes {
		delete(i.hodlSubscriptions[hash], subscriber)
	}

	delete(i.hodlReverseSubscriptions, subscriber)
}
912579
func (t *task) SubscribePlugins() ([]string, []serror.SnapError) {
	depGroups := getWorkflowPlugins(t.workflow.processNodes, t.workflow.publishNodes, t.workflow.metrics)
	var subbedDeps []string
	for k := range depGroups {
		var errs []serror.SnapError
		mgr, err := t.RemoteManagers.Get(k)
		if err != nil {
			errs = append(errs, serror.New(err))
		} else {
			errs = mgr.SubscribeDeps(t.ID(), depGroups[k].requestedMetrics, depGroups[k].subscribedPlugins, t.workflow.configTree)
		}
		// If there are errors with subscribing any deps, go through and unsubscribe all other
		// deps that may have already been subscribed then return the errors.
		if len(errs) > 0 {
			for _, key := range subbedDeps {
				mgr, err := t.RemoteManagers.Get(key)
				if err != nil {
					errs = append(errs, serror.New(err))
				} else {
					// sending empty mts to unsubscribe to indicate task should not start
					uerrs := mgr.UnsubscribeDeps(t.ID())
					errs = append(errs, uerrs...)
				}
			}
			return nil, errs
		}
		// If subscribed successfully add to subbedDeps
		subbedDeps = append(subbedDeps, k)
	}

	return subbedDeps, nil
}