// Remove consumer subscription on given topic. If there is no such a subscription, return error.
188779
func (mux *Multiplexer) stopConsuming(topic string, name string) error {
	mux.rwlock.Lock()
	defer mux.rwlock.Unlock()

	var wasError error
	var topicFound bool
	for index, subs := range mux.mapping {
		if !subs.manual && subs.topic == topic && subs.connectionName == name {
			topicFound = true
			mux.mapping = append(mux.mapping[:index], mux.mapping[index+1:]...)
		}
	}
	if !topicFound {
		wasError = fmt.Errorf("topic %s was not consumed by '%s'", topic, name)
	}
	return wasError
}
581446
func (u *UserResource) removeUser(request *restful.Request, response *restful.Response) {
	id := request.PathParameter("user-id")
	delete(u.users, id)
}
1282583
func (em *EventMeter) Unsubscribe(query string) error {
	em.mtx.Lock()
	defer em.mtx.Unlock()

	return em.wsc.Unsubscribe(context.TODO(), query)
}
1072857
func (s *Service) RemoveService(w http.ResponseWriter, r *http.Request) {
	id, err := paramID("kid", r)
	if err != nil {
		Error(w, http.StatusUnprocessableEntity, err.Error(), s.Logger)
		return
	}

	srcID, err := paramID("id", r)
	if err != nil {
		Error(w, http.StatusUnprocessableEntity, err.Error(), s.Logger)
		return
	}

	ctx := r.Context()
	srv, err := s.Store.Servers(ctx).Get(ctx, id)
	if err != nil || srv.SrcID != srcID || srv.Type == "" {
		notFound(w, id, s.Logger)
		return
	}

	if err = s.Store.Servers(ctx).Delete(ctx, srv); err != nil {
		unknownErrorWithMessage(w, err, s.Logger)
		return
	}

	w.WriteHeader(http.StatusNoContent)
}
784595
func (r *RoleV2) RemoveResource(kind string) {
	delete(r.Spec.Resources, kind)
}
147617
func (em *clientEventMonitor) unsubscribe(s *Subscription) error {
	em.Lock()
	defer em.Unlock()

	ns := []*Subscription{}
	for _, sub := range em.subscriptions[s.ID] {
		if sub != s {
			ns = append(ns, sub)
		}
	}

	em.subscriptions[s.ID] = ns

	return nil
}
234948
func (c *Cluster) RemoveService(input string) error {
	return c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
		service, err := getService(ctx, state.controlClient, input, false)
		if err != nil {
			return err
		}

		_, err = state.controlClient.RemoveService(ctx, &swarmapi.RemoveServiceRequest{ServiceID: service.ID})
		return err
	})
}
1205051
func (s *StanServer) removeAllNonDurableSubscribers(client *client) {
	// client has been unregistered and no other routine can add/remove
	// subscriptions, so it is safe to use the original.
	client.RLock()
	subs := client.subs
	clientID := client.info.ID
	client.RUnlock()
	var (
		storesToFlush = map[string]stores.SubStore{}
		channels      = map[string]struct{}{}
	)
	for _, sub := range subs {
		sub.RLock()
		subject := sub.subject
		isDurable := sub.IsDurable
		subStore := sub.store
		sub.RUnlock()
		// Get the channel
		c := s.channels.get(subject)
		if c == nil {
			continue
		}
		// Don't remove durables
		c.ss.Remove(c, sub, false)
		// If the sub is a durable, there may have been an update to storage,
		// so we will want to flush the store. In clustering, during replay,
		// subStore may be nil.
		if isDurable && subStore != nil {
			storesToFlush[subject] = subStore
		}
		channels[subject] = struct{}{}
	}
	if len(storesToFlush) > 0 {
		for subject, subStore := range storesToFlush {
			if err := subStore.Flush(); err != nil {
				s.log.Errorf("[Client:%s] Error flushing store while removing subscriptions: subject=%s, err=%v", clientID, subject, err)
			}
		}
	}
	for channel := range channels {
		s.channels.maybeStartChannelDeleteTimer(channel, nil)
	}
}
132709
func NewConsumer(topicName string, subscriptionName string) Consumer {
	pubsubClient, err := newPubSubClient()

	if err != nil {
		log.Fatalf("Could not create PubSub client: %v", err)
	}

	topic := ensureTopic(pubsubClient, topicName)
	sub := ensureSubscription(pubsubClient, topic, subscriptionName)

	return &googlePubSubConsumer{Subscription: sub}
}
439884
func (m *MemoryBackend) Unsubscribe(client *Client, topics []string, ack Ack) error {
	// delete subscriptions
	for _, t := range topics {
		client.Session().(*memorySession).subscriptions.Empty(t)
	}

	// call ack if provided
	if ack != nil {
		ack()
	}

	return nil
}
734373
func (s *Sublist) remove(sub *subscription, shouldLock bool) error {
	subject := string(sub.subject)
	tsa := [32]string{}
	tokens := tsa[:0]
	start := 0
	for i := 0; i < len(subject); i++ {
		if subject[i] == btsep {
			tokens = append(tokens, subject[start:i])
			start = i + 1
		}
	}
	tokens = append(tokens, subject[start:])

	if shouldLock {
		s.Lock()
		defer s.Unlock()
	}

	sfwc := false
	l := s.root
	var n *node

	// Track levels for pruning
	var lnts [32]lnt
	levels := lnts[:0]

	for _, t := range tokens {
		lt := len(t)
		if lt == 0 || sfwc {
			return ErrInvalidSubject
		}
		if l == nil {
			return ErrNotFound
		}
		if lt > 1 {
			n = l.nodes[t]
		} else {
			switch t[0] {
			case pwc:
				n = l.pwc
			case fwc:
				n = l.fwc
				sfwc = true
			default:
				n = l.nodes[t]
			}
		}
		if n != nil {
			levels = append(levels, lnt{l, n, t})
			l = n.next
		} else {
			l = nil
		}
	}
	if !s.removeFromNode(n, sub) {
		return ErrNotFound
	}

	s.count--
	s.removes++

	for i := len(levels) - 1; i >= 0; i-- {
		l, n, t := levels[i].l, levels[i].n, levels[i].t
		if n.isEmpty() {
			l.pruneNode(n, t)
		}
	}
	s.removeFromCache(subject, sub)
	atomic.AddUint64(&s.genid, 1)

	return nil
}
1243745
func (client *BaseClient) RemoveFilterByIndex(index int) Client {
	client.filterManager.RemoveFilterByIndex(index)
	return client
}
1248321
func (c *DefaultClient) UnsubscribeAppEvents(appID string, eventType types.EventType) Token {
	topic := ApplicationTopic{appID, AppEvents, string(eventType)}
	return c.unsubscribe(topic.String())
}
582186
func (exchange *Exchange) Unsubscribe(handle Handle) error {
	exchange.cond.L.Lock()
	defer exchange.cond.L.Unlock()

	if exchange.state != stateRunning {
		return ErrInvalidState
	}

	topic, ok := exchange.topicForHandle[handle]
	if !ok {
		return ErrInvalidHandle
	}

	delete(exchange.topicForHandle, handle)

	item := exchange.trie.Get(patricia.Prefix(topic))
	if item == nil {
		return ErrInvalidHandle
	}

	records := item.(*[]*handlerRecord)
	if len(*records) == 1 {
		if exchange.trie.Delete(patricia.Prefix(topic)) {
			return nil
		}
	} else {
		for i, record := range *records {
			if record.handle == handle {
				*records = append((*records)[:i], (*records)[i+1:]...)
				return nil
			}
		}
	}

	return ErrInvalidHandle
}
1189489
func (s *ticketStore) removeRegisterTopic(topic Topic) {
	log.Trace("Removing discovery topic", "topic", topic)
	if s.tickets[topic] == nil {
		log.Warn("Removing non-existent discovery topic", "topic", topic)
		return
	}
	for _, list := range s.tickets[topic].buckets {
		for _, ref := range list {
			ref.t.refCnt--
			if ref.t.refCnt == 0 {
				delete(s.nodes, ref.t.node)
				delete(s.nodeLastReq, ref.t.node)
			}
		}
	}
	delete(s.tickets, topic)
}
411107
func (c *Client) Unsubscribe(topic string) (GenericFuture, error) {
	return c.UnsubscribeMultiple([]string{topic})
}
450425
func (s *Sublist) removeFromCache(subject string, sub *subscription) {
	if s.cache == nil {
		return
	}
	// If literal we can direct match.
	if subjectIsLiteral(subject) {
		// Load for accounting
		if _, ok := s.cache.Load(subject); ok {
			s.cache.Delete(subject)
			atomic.AddInt32(&s.cacheNum, -1)
		}
		return
	}
	s.cache.Range(func(k, v interface{}) bool {
		key := k.(string)
		if matchLiteral(key, subject) {
			// Since someone else may be referecing, can't modify the list
			// safely, just let it re-populate.
			s.cache.Delete(key)
			atomic.AddInt32(&s.cacheNum, -1)
		}
		return true
	})
}