// NewSubscription returns a new subscription with the given outCapacity.
326046
func NewSubscription(outCapacity int) *Subscription {
	return &Subscription{
		out:       make(chan Message, outCapacity),
		cancelled: make(chan struct{}),
	}
}
623754
func (gss *genericSubStore) createSub(sub *spb.SubState) error {
	if gss.limits.MaxSubscriptions > 0 && len(gss.subs) >= gss.limits.MaxSubscriptions {
		return ErrTooManySubs
	}

	// Bump the max value before assigning it to the new subscription.
	gss.maxSubID++

	// This new subscription has the max value.
	sub.ID = gss.maxSubID
	// Store anything. Some implementations may replace with specific
	// object.
	gss.subs[sub.ID] = emptySub

	return nil
}
21943
func NewSubscription(producer func(<-chan struct{}) error) Subscription {
	s := &funcSub{unsub: make(chan struct{}), err: make(chan error, 1)}
	go func() {
		defer close(s.err)
		err := producer(s.unsub)
		s.mu.Lock()
		defer s.mu.Unlock()
		if !s.unsubscribed {
			if err != nil {
				s.err <- err
			}
			s.unsubscribed = true
		}
	}()
	return s
}
863780
func SubscriberContext(ctx context.Context) SubscriberOption {
	return func(o *SubscriberOptions) {
		o.Context = ctx
	}
}
1052589
func (s *DescribeHsmOutput) SetSubscriptionType(v string) *DescribeHsmOutput {
	s.SubscriptionType = &v
	return s
}
917741
func newSubquery(alias sqlparser.TableIdent, bldr builder) (*subquery, *symtab, error) {
	sq := &subquery{
		order:     bldr.Order() + 1,
		input:     bldr,
		esubquery: &engine.Subquery{},
	}

	// Create a 'table' that represents the subquery.
	t := &table{
		alias:  sqlparser.TableName{Name: alias},
		origin: sq,
	}

	// Create column symbols based on the result column names.
	for _, rc := range bldr.ResultColumns() {
		if _, ok := t.columns[rc.alias.Lowered()]; ok {
			return nil, nil, fmt.Errorf("duplicate column names in subquery: %s", sqlparser.String(rc.alias))
		}
		t.addColumn(rc.alias, &column{origin: sq})
	}
	t.isAuthoritative = true
	st := newSymtab()
	// AddTable will not fail because symtab is empty.
	_ = st.AddTable(t)
	return sq, st, nil
}
309609
func NewSubscriptionManager(ntfnSource NotificationSource) *SubscriptionManager {
	return &SubscriptionManager{
		subscribers:         make(map[uint64]*newSubscription),
		newSubscriptions:    make(chan *newSubscription),
		cancelSubscriptions: make(chan *cancelSubscription),
		ntfnSource:          ntfnSource,
		quit:                make(chan struct{}),
	}
}
313047
func NewSubnetAPI(client *Client) *SubnetAPI {
	return &SubnetAPI{
		&baseAPI{
			client: client,
			FuncGetResourceURL: func() string {
				return "subnet"
			},
		},
	}
}
244003
func (t *TrackingSettings) SetSubscriptionTracking(subscriptionTracking *SubscriptionTrackingSetting) *TrackingSettings {
	t.SubscriptionTracking = subscriptionTracking
	return t
}
667031
func (s *Server) sysSubscribe(subject string, cb msgHandler) (*subscription, error) {
	if !s.eventsEnabled() {
		return nil, ErrNoSysAccount
	}
	if cb == nil {
		return nil, fmt.Errorf("undefined message handler")
	}
	s.mu.Lock()
	sid := strconv.FormatInt(int64(s.sys.sid), 10)
	s.sys.subs[sid] = cb
	s.sys.sid++
	c := s.sys.client
	s.mu.Unlock()

	// Now create the subscription
	if err := c.processSub([]byte(subject + " " + sid)); err != nil {
		return nil, err
	}
	c.mu.Lock()
	sub := c.subs[sid]
	c.mu.Unlock()
	return sub, nil
}
1078400
func NewSubnets(ipNet *net.IPNet) (Subnets, error) {
	return &pool{dynamicRange: ipNet, allocated: make(map[string][]net.IP)}, nil
}
356561
func NewSubscriberWrapper() server.SubscriberWrapper {
	return func(fn server.SubscriberFunc) server.SubscriberFunc {
		return func(ctx context.Context, p server.Message) (err error) {
			t := newPublicationTracker(p, ServerProfile)
			ctx = t.start(ctx, false)

			defer func() { t.end(ctx, err) }()

			spanCtx := getTraceFromCtx(ctx)
			if spanCtx != nil {
				ctx, t.span = trace.StartSpanWithRemoteParent(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
					*spanCtx,
				)
			} else {
				ctx, t.span = trace.StartSpan(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
				)
			}

			err = fn(ctx, p)
			return
		}
	}
}
335763
func CreateSubscription(b bool) broker.Option {
	return func(o *broker.Options) {
		if o.Context == nil {
			o.Context = context.Background()
		}

		o.Context = context.WithValue(o.Context, createSubscription{}, b)
	}
}
687057
func (b *pubsubBroker) Subscribe(topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	options := broker.SubscribeOptions{
		AutoAck: true,
		Queue:   "q-" + uuid.New().String(),
		Context: b.options.Context,
	}

	for _, o := range opts {
		o(&options)
	}

	ctx := context.Background()
	sub := b.client.Subscription(options.Queue)

	if createSubscription, ok := b.options.Context.Value(createSubscription{}).(bool); !ok || createSubscription {
		exists, err := sub.Exists(ctx)
		if err != nil {
			return nil, err
		}

		if !exists {
			tt := b.client.Topic(topic)
			subb, err := b.client.CreateSubscription(ctx, options.Queue, pubsub.SubscriptionConfig{
				Topic:       tt,
				AckDeadline: time.Duration(0),
			})
			if err != nil {
				return nil, err
			}
			sub = subb
		}
	}

	subscriber := &subscriber{
		options: options,
		topic:   topic,
		exit:    make(chan bool),
		sub:     sub,
	}

	go subscriber.run(h)

	return subscriber, nil
}
871817
func (sub *Subscription) subUnsub(cmd string, channels ...string) error {
	err := sub.ensureProtocol()
	if err != nil {
		return err
	}
	pattern := false
	args := []interface{}{}
	for _, channel := range channels {
		if containsPattern(channel) {
			pattern = true
		}
		args = append(args, channel)
	}
	if pattern {
		cmd = "p" + cmd
	}
	err = sub.resp.sendCommand(cmd, args...)
	logCommand(cmd, args, err, sub.database.logging)
	return err
}
239081
func NewSubnet(a Address, m AddressMask) (Subnet, error) {
	if len(a) != len(m) {
		return Subnet{}, errSubnetLengthMismatch
	}
	for i := 0; i < len(a); i++ {
		if a[i]&^m[i] != 0 {
			return Subnet{}, errSubnetAddressMasked
		}
	}
	return Subnet{a, m}, nil
}
628157
func (s *SubscriptionProvider) Subscribe() *Subscription {
	s.Lock()
	defer s.Unlock()
	if s.subscriptions == nil {
		s.subscriptions = make(map[string]*Subscription)
	}
	sub := newSubscription(s)
	s.subscriptions[sub.id.String()] = sub

	return sub
}