func (s *Subscription) Up() <-chan *ContextualApplicationUp {
	return s.upCh
}
Up returns the upstream channel.
package io
import (
	"context"
	"fmt"
	"net/http"
	"go.thethings.network/lorawan-stack/v3/pkg/cluster"
	"go.thethings.network/lorawan-stack/v3/pkg/component"
	"go.thethings.network/lorawan-stack/v3/pkg/config"
	"go.thethings.network/lorawan-stack/v3/pkg/errorcontext"
	"go.thethings.network/lorawan-stack/v3/pkg/errors"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
	"go.thethings.network/lorawan-stack/v3/pkg/ratelimit"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
	"google.golang.org/grpc"
)
const DefaultBufferSize = 128
type PubSub interface {
	Publish(ctx context.Context, up *ttnpb.ApplicationUp) error
	Subscribe(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, cluster bool) (*Subscription, error)
}
type DownlinkQueueOperator interface {
	DownlinkQueuePush(context.Context, ttnpb.EndDeviceIdentifiers, []*ttnpb.ApplicationDownlink) error
	DownlinkQueueReplace(context.Context, ttnpb.EndDeviceIdentifiers, []*ttnpb.ApplicationDownlink) error
	DownlinkQueueList(context.Context, ttnpb.EndDeviceIdentifiers) ([]*ttnpb.ApplicationDownlink, error)
}
type UplinkStorage interface {
	RangeUplinks(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, paths []string, f func(ctx context.Context, up *ttnpb.ApplicationUplink) bool) error
}
type Cluster interface {
	GetPeers(ctx context.Context, role ttnpb.ClusterRole) ([]cluster.Peer, error)
	GetPeer(ctx context.Context, role ttnpb.ClusterRole, ids cluster.EntityIdentifiers) (cluster.Peer, error)
	GetPeerConn(ctx context.Context, role ttnpb.ClusterRole, ids cluster.EntityIdentifiers) (*grpc.ClientConn, error)
}
type Server interface {
	component.TaskStarter
	PubSub
	DownlinkQueueOperator
	UplinkStorage
	Cluster
	FromRequestContext(context.Context) context.Context
	GetBaseConfig(ctx context.Context) config.ServiceBase
	FillContext(ctx context.Context) context.Context
	HTTPClient(context.Context) (*http.Client, error)
	RateLimiter() ratelimit.Interface
}
type ContextualApplicationUp struct {
	context.Context
	*ttnpb.ApplicationUp
}
type Subscription struct {
	ctx       context.Context
	cancelCtx errorcontext.CancelFunc
	protocol string
	ids      *ttnpb.ApplicationIdentifiers
	upCh    chan *ContextualApplicationUp
	publish func(context.Context, context.Context, chan<- *ContextualApplicationUp, *ContextualApplicationUp) error
}
type SubscriptionOption interface {
	apply(*Subscription)
}
type subscriptionOptionFunc func(s *Subscription)
func (f subscriptionOptionFunc) apply(s *Subscription) { f(s) }
func WithBlocking(blocking bool) SubscriptionOption {
	return subscriptionOptionFunc(func(s *Subscription) {
		if blocking {
			s.publish = blockingPublish
		} else {
			s.publish = nonBlockingPublish
		}
	})
}
func WithBufferSize(bufferSize int) SubscriptionOption {
	return subscriptionOptionFunc(func(s *Subscription) {
		s.upCh = make(chan *ContextualApplicationUp, bufferSize)
	})
}
func NewSubscription(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, opts ...SubscriptionOption) *Subscription {
	ctx, cancelCtx := errorcontext.New(ctx)
	s := &Subscription{
		ctx:       ctx,
		cancelCtx: cancelCtx,
		protocol:  protocol,
		ids:       ids,
		upCh:      make(chan *ContextualApplicationUp, DefaultBufferSize),
		publish:   nonBlockingPublish,
	}
	for _, opt := range opts {
		opt.apply(s)
	}
	return s
}
func (s *Subscription) Context() context.Context { return s.ctx }
func (s *Subscription) Disconnect(err error) {
	s.cancelCtx(err)
}
func (s *Subscription) Protocol() string { return s.protocol }
func (s *Subscription) ApplicationIDs() *ttnpb.ApplicationIdentifiers { return s.ids }
func (s *Subscription) Publish(ctx context.Context, up *ttnpb.ApplicationUp) error {
	ctxUp := &ContextualApplicationUp{
		Context:       ctx,
		ApplicationUp: up,
	}
	return s.publish(ctx, s.ctx, s.upCh, ctxUp)
}
func blockingPublish(ctx context.Context, subCtx context.Context, upCh chan<- *ContextualApplicationUp, up *ContextualApplicationUp) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-subCtx.Done():
		return subCtx.Err()
	case upCh <- up:
		return nil
	}
}
var errBufferFull = errors.DefineResourceExhausted("buffer_full", "buffer is full")
func nonBlockingPublish(ctx context.Context, subCtx context.Context, upCh chan<- *ContextualApplicationUp, up *ContextualApplicationUp) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-subCtx.Done():
		return subCtx.Err()
	case upCh <- up:
		return nil
	default:
		return errBufferFull.New()
	}
}