// UnsubscribePlugins groups task dependencies by the node they live in workflow and unsubscribe them
426974
func (t *task) UnsubscribePlugins() []serror.SnapError {
	depGroups := getWorkflowPlugins(t.workflow.processNodes, t.workflow.publishNodes, t.workflow.metrics)
	var errs []serror.SnapError
	for k := range depGroups {
		event := &scheduler_event.PluginsUnsubscribedEvent{
			TaskID:  t.ID(),
			Plugins: depGroups[k].subscribedPlugins,
		}
		defer t.eventEmitter.Emit(event)
		mgr, err := t.RemoteManagers.Get(k)
		if err != nil {
			errs = append(errs, serror.New(err))
		} else {
			uerrs := mgr.UnsubscribeDeps(t.ID())
			if len(uerrs) > 0 {
				errs = append(errs, uerrs...)
			}
		}
	}
	for _, err := range errs {
		taskLogger.WithFields(log.Fields{
			"_block":     "UnsubscribePlugins",
			"task-id":    t.id,
			"task-name":  t.name,
			"task-state": t.state,
		}).Error(err)
	}
	return errs
}
485266
func (d *IdentifyDispatch) Subscribe(mctx MetaContext) (unsubscribe func(), recvCh <-chan IdentifyDispatchMsg) {
	mctx.Debug("IdentifyDispatch.Subscribe")
	ch := make(chan IdentifyDispatchMsg, 10)
	d.Lock()
	defer d.Unlock()
	d.listeners = append(d.listeners, ch)
	unsubscribe = func() {
		mctx.Debug("IdentifyDispatch.Unsubcribe")
		d.Lock()
		defer d.Unlock()
		var listeners []chan<- IdentifyDispatchMsg
		for _, ch2 := range d.listeners {
			if ch == ch2 {
				continue
			}
			listeners = append(listeners, ch2)
		}
		d.listeners = listeners
	}
	return unsubscribe, ch
}
85111
func NewUNSUBSCRIBE(opts *UNSUBSCRIBEOptions) (Packet, error) {
	// Initialize the options.
	if opts == nil {
		opts = &UNSUBSCRIBEOptions{}
	}

	// Validate the options.
	if err := opts.validate(); err != nil {
		return nil, err
	}

	// Create a SUBSCRIBE Packet.
	p := &UNSUBSCRIBE{
		PacketID:     opts.PacketID,
		TopicFilters: opts.TopicFilters,
	}

	// Set the variable header to the Packet.
	p.setVariableHeader()

	// Set the payload to the Packet.
	p.setPayload()

	// Set the Fixed header to the Packet.
	p.setFixedHeader()

	// Return the Packet.
	return p, nil
}
850217
func (s *Subscription) Unsubscribe() error {
	if s == nil {
		return ErrBadSubscription
	}
	s.mu.Lock()
	conn := s.conn
	s.mu.Unlock()
	if conn == nil {
		return ErrBadSubscription
	}
	if conn.IsDraining() {
		return ErrConnectionDraining
	}
	return conn.unsubscribe(s, 0, false)
}
411107
func (c *Client) Unsubscribe(topic string) (GenericFuture, error) {
	return c.UnsubscribeMultiple([]string{topic})
}
603219
func (p *Peer) Unsubscribe(sub peer.Subscriber) error {
	p.lock.Lock()
	defer p.lock.Unlock()
	if _, ok := p.subscribers[sub]; !ok {
		return peer.ErrPeerHasNoReferenceToSubscriber{
			PeerIdentifier: p.PeerIdentifier,
			PeerSubscriber: sub,
		}
	}

	delete(p.subscribers, sub)
	return nil
}
871468
func (mr *MockAgentConnectorMockRecorder) Unsubscribe(name interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unsubscribe", reflect.TypeOf((*MockAgentConnector)(nil).Unsubscribe), name)
}
1267009
func (n *Nats) UnsubscribeAll() {
	for _, route := range n.Opts.routes {
		route.subsc.Unsubscribe()
	}
}
219690
func (nc *Conn) unsubscribe(sub *Subscription, max int, drainMode bool) error {
	nc.mu.Lock()
	// ok here, but defer is expensive
	defer nc.mu.Unlock()
	defer nc.kickFlusher()

	if nc.isClosed() {
		return ErrConnectionClosed
	}

	nc.subsMu.RLock()
	s := nc.subs[sub.sid]
	nc.subsMu.RUnlock()
	// Already unsubscribed
	if s == nil {
		return nil
	}

	maxStr := _EMPTY_
	if max > 0 {
		s.max = uint64(max)
		maxStr = strconv.Itoa(max)
	} else if !drainMode {
		nc.removeSub(s)
	}

	if drainMode {
		go nc.checkDrained(sub)
	}

	// We will send these for all subs when we reconnect
	// so that we can suppress here.
	if !nc.isReconnecting() {
		fmt.Fprintf(nc.bw, unsubProto, s.sid, maxStr)
	}
	return nil
}
748560
func (c PubSubConn) Unsubscribe(channel ...interface{}) error {
	c.Conn.Send("UNSUBSCRIBE", channel...)
	return c.Conn.Flush()
}
1248321
func (c *DefaultClient) UnsubscribeAppEvents(appID string, eventType types.EventType) Token {
	topic := ApplicationTopic{appID, AppEvents, string(eventType)}
	return c.unsubscribe(topic.String())
}
737921
func (em *Emitter) UnsubscribeAll(ctx context.Context, subscriber string) error {
	return em.pubsubServer.UnsubscribeAll(ctx, subscriber)
}
302754
func (r *marathonClient) Unsubscribe(callback string) error {
	// step: remove from the list of subscriptions
	return r.apiDelete(fmt.Sprintf("%s?callbackUrl=%s", marathonAPISubscription, callback), nil, nil)
}
259951
func (mg *MailgunImpl) DeleteUnsubscribe(ctx context.Context, address string) error {
	r := newHTTPRequest(generateApiUrlWithTarget(mg, unsubscribesEndpoint, address))
	r.setClient(mg.Client())
	r.setBasicAuth(basicAuthUser, mg.APIKey())
	_, err := makeDeleteRequest(ctx, r)
	return err
}
281795
func (c *conn) unsubscribed(char *ble.Characteristic) {
	if n, found := c.notifiers[char.Handle]; found {
		n.Close()
		delete(c.notifiers, char.Handle)
	}
}
1138549
func ToCorePlugins(pl []core.SubscribedPlugin) []core.Plugin {
	plugins := make([]core.Plugin, len(pl))
	for i, v := range pl {
		plugins[i] = v
	}
	return plugins
}
43031
func (c *Client) Plugins() ([]string, error) {
	stats, err := c.ClusterStats().Do(context.Background())
	if err != nil {
		return nil, err
	}
	if stats == nil {
		return nil, err
	}
	if stats.Nodes == nil {
		return nil, err
	}
	var plugins []string
	for _, plugin := range stats.Nodes.Plugins {
		plugins = append(plugins, plugin.Name)
	}
	return plugins, nil
}