func (a *Agent) RegisterHandlers(s *runtime.ServeMux, conn *grpc.ClientConn) {
	ttnpb.RegisterPbaHandler(a.Context(), s, conn)
}
RegisterHandlers registers gRPC handlers.
package packetbrokeragent
import (
	"context"
	"encoding/hex"
	"fmt"
	"sync"
	"time"
	"github.com/grpc-ecosystem/grpc-gateway/runtime"
	routingpb "go.packetbroker.org/api/routing"
	packetbroker "go.packetbroker.org/api/v3"
	"go.thethings.network/lorawan-stack/v3/pkg/cluster"
	"go.thethings.network/lorawan-stack/v3/pkg/component"
	"go.thethings.network/lorawan-stack/v3/pkg/encoding/lorawan"
	"go.thethings.network/lorawan-stack/v3/pkg/errors"
	"go.thethings.network/lorawan-stack/v3/pkg/events"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
	"go.thethings.network/lorawan-stack/v3/pkg/random"
	"go.thethings.network/lorawan-stack/v3/pkg/rpcclient"
	"go.thethings.network/lorawan-stack/v3/pkg/rpcmiddleware/hooks"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
	"go.thethings.network/lorawan-stack/v3/pkg/types"
	"go.thethings.network/lorawan-stack/v3/pkg/unique"
	"go.thethings.network/lorawan-stack/v3/pkg/workerpool"
	"google.golang.org/grpc"
	"gopkg.in/square/go-jose.v2"
)
const (
	subscribeStreamCount = 8
	publishMessageTimeout = 3 * time.Second
)
type TenantContextFiller func(parent context.Context, tenantID string) (context.Context, error)
type TenantIDExtractor func(ctx context.Context) string
type RegistrationInfo struct {
	Name          string
	DevAddrBlocks []*ttnpb.PacketBrokerDevAddrBlock
	ContactInfo   []*ttnpb.ContactInfo
	Listed        bool
}
type RegistrationInfoExtractor func(ctx context.Context, homeNetworkClusterID string) (*RegistrationInfo, error)
type uplinkMessage struct {
	context.Context
	*packetbroker.UplinkMessage
}
type downlinkMessage struct {
	context.Context
	*agentUplinkToken
	*packetbroker.DownlinkMessage
}
type Agent struct {
	*component.Component
	ctx context.Context
	dataPlaneAddress string
	netID            types.NetID
	subscriptionTenantID,
	clusterID,
	homeNetworkClusterID string
	authenticator     authenticator
	forwarderConfig   ForwarderConfig
	homeNetworkConfig HomeNetworkConfig
	devAddrPrefixes   []types.DevAddrPrefix
	tenantContextFillers      []TenantContextFiller
	tenantIDExtractor         TenantIDExtractor
	registrationInfoExtractor RegistrationInfoExtractor
	upstreamCh   chan *uplinkMessage
	downstreamCh chan *downlinkMessage
	grpc struct {
		pba   ttnpb.PbaServer
		nsPba ttnpb.NsPbaServer
		gsPba ttnpb.GsPbaServer
	}
}
type Option func(*Agent)
func WithTenantContextFiller(filler TenantContextFiller) Option {
	return func(a *Agent) {
		a.tenantContextFillers = append(a.tenantContextFillers, filler)
	}
}
func WithTenantIDExtractor(extractor TenantIDExtractor) Option {
	return func(a *Agent) {
		a.tenantIDExtractor = extractor
	}
}
func WithRegistrationInfo(extractor RegistrationInfoExtractor) Option {
	return func(a *Agent) {
		a.registrationInfoExtractor = extractor
	}
}
var (
	errNetID              = errors.DefineFailedPrecondition("net_id", "invalid NetID `{net_id}`")
	errAuthenticationMode = errors.DefineFailedPrecondition("authentication_mode", "invalid authentication mode `{mode}`")
	errTokenKey           = errors.DefineFailedPrecondition("token_key", "invalid token key", "length")
)
func New(c *component.Component, conf *Config, opts ...Option) (*Agent, error) {
	ctx := log.NewContextWithField(c.Context(), "namespace", "packetbrokeragent")
	logger := log.FromContext(ctx)
	var devAddrPrefixes []types.DevAddrPrefix
	if conf.HomeNetwork.Enable {
		devAddrPrefixes = append(devAddrPrefixes, conf.HomeNetwork.DevAddrPrefixes...)
		if len(devAddrPrefixes) == 0 {
			devAddr, err := types.NewDevAddr(conf.NetID, nil)
			if err != nil {
				return nil, errNetID.WithAttributes("net_id", conf.NetID).WithCause(err)
			}
			devAddrPrefix := types.DevAddrPrefix{
				DevAddr: devAddr,
				Length:  uint8(conf.NetID.IDBits()),
			}
			devAddrPrefixes = append(devAddrPrefixes, devAddrPrefix)
		}
	}
	var authenticator authenticator
	switch mode := conf.AuthenticationMode; mode {
	case "oauth2":
		var tlsConfig tlsConfigurator
		if !conf.Insecure {
			tlsConfig = c
		}
		authenticator = newOAuth2(ctx, conf.OAuth2, tlsConfig,
			conf.IAMAddress,
			conf.ControlPlaneAddress,
			conf.DataPlaneAddress,
			conf.MapperAddress,
		)
	}
	homeNetworkClusterID := conf.HomeNetworkClusterID
	if homeNetworkClusterID == "" {
		homeNetworkClusterID = conf.ClusterID
	}
	a := &Agent{
		Component:            c,
		ctx:                  ctx,
		dataPlaneAddress:     conf.DataPlaneAddress,
		netID:                conf.NetID,
		subscriptionTenantID: conf.TenantID,
		clusterID:            conf.ClusterID,
		homeNetworkClusterID: homeNetworkClusterID,
		authenticator:        authenticator,
		forwarderConfig:      conf.Forwarder,
		homeNetworkConfig:    conf.HomeNetwork,
		devAddrPrefixes:      devAddrPrefixes,
		tenantIDExtractor: func(_ context.Context) string {
			return conf.TenantID
		},
		registrationInfoExtractor: func(_ context.Context, homeNetworkClusterID string) (*RegistrationInfo, error) {
			blocks := make([]*ttnpb.PacketBrokerDevAddrBlock, len(devAddrPrefixes))
			for i, p := range devAddrPrefixes {
				blocks[i] = &ttnpb.PacketBrokerDevAddrBlock{
					DevAddrPrefix: &ttnpb.DevAddrPrefix{
						DevAddr: &p.DevAddr,
						Length:  uint32(p.Length),
					},
					HomeNetworkClusterId: homeNetworkClusterID,
				}
			}
			contactInfo := make([]*ttnpb.ContactInfo, 0, 2)
			if adminContact := conf.Registration.AdministrativeContact.ContactInfo(ttnpb.CONTACT_TYPE_OTHER); adminContact != nil {
				contactInfo = append(contactInfo, adminContact)
			}
			if techContact := conf.Registration.TechnicalContact.ContactInfo(ttnpb.CONTACT_TYPE_TECHNICAL); techContact != nil {
				contactInfo = append(contactInfo, techContact)
			}
			return &RegistrationInfo{
				Name:          conf.Registration.Name,
				DevAddrBlocks: blocks,
				ContactInfo:   contactInfo,
				Listed:        conf.Registration.Listed,
			}, nil
		},
	}
	if a.forwarderConfig.Enable {
		a.upstreamCh = make(chan *uplinkMessage)
		if a.forwarderConfig.WorkerPool.Limit <= 1 {
			a.forwarderConfig.WorkerPool.Limit = 2
		}
		if len(a.forwarderConfig.TokenKey) == 0 {
			a.forwarderConfig.TokenKey = random.Bytes(16)
			logger.WithField("token_key", hex.EncodeToString(a.forwarderConfig.TokenKey)).Warn("No token key configured, generated a random one")
		}
		var (
			enc jose.ContentEncryption
			alg jose.KeyAlgorithm
		)
		switch l := len(a.forwarderConfig.TokenKey); l {
		case 16:
			enc, alg = jose.A128GCM, jose.A128GCMKW
		case 32:
			enc, alg = jose.A256GCM, jose.A256GCMKW
		default:
			return nil, errTokenKey.WithAttributes("length", l).New()
		}
		var err error
		a.forwarderConfig.TokenEncrypter, err = jose.NewEncrypter(enc, jose.Recipient{
			Algorithm: alg,
			Key:       a.forwarderConfig.TokenKey,
		}, nil)
		if err != nil {
			return nil, errTokenKey.WithCause(err)
		}
	}
	if a.homeNetworkConfig.Enable {
		a.downstreamCh = make(chan *downlinkMessage)
		if a.homeNetworkConfig.WorkerPool.Limit <= 1 {
			a.homeNetworkConfig.WorkerPool.Limit = 2
		}
	}
	for _, opt := range opts {
		opt(a)
	}
	if a.authenticator == nil {
		return nil, errAuthenticationMode.WithAttributes("mode", conf.AuthenticationMode)
	}
	if a.forwarderConfig.Enable || a.homeNetworkConfig.Enable {
		iamConn, err := a.dialContext(ctx, conf.IAMAddress)
		if err != nil {
			return nil, err
		}
		cpConn, err := a.dialContext(ctx, conf.ControlPlaneAddress)
		if err != nil {
			return nil, err
		}
		a.grpc.pba = &pbaServer{
			Agent:   a,
			iamConn: iamConn,
			cpConn:  cpConn,
		}
	} else {
		a.grpc.pba = &disabledServer{}
	}
	if a.forwarderConfig.Enable {
		mapperConn, err := a.dialContext(ctx, conf.MapperAddress)
		if err != nil {
			return nil, err
		}
		a.grpc.gsPba = &gsPbaServer{
			netID:               a.netID,
			clusterID:           a.clusterID,
			config:              a.forwarderConfig,
			messageEncrypter:    a,
			contextDecoupler:    a,
			tenantIDExtractor:   a.tenantIDExtractor,
			frequencyPlansStore: a.FrequencyPlans,
			upstreamCh:          a.upstreamCh,
			mapperConn:          mapperConn,
		}
	} else {
		a.grpc.gsPba = &disabledServer{}
	}
	if a.homeNetworkConfig.Enable {
		a.grpc.nsPba = &nsPbaServer{
			contextDecoupler: a,
			downstreamCh:     a.downstreamCh,
			frequencyPlans:   a.FrequencyPlans,
		}
	} else {
		a.grpc.nsPba = &disabledServer{}
	}
	newTaskConfig := func(id string, fn component.TaskFunc) *component.TaskConfig {
		return &component.TaskConfig{
			Context: ctx,
			ID:      id,
			Func:    fn,
			Restart: component.TaskRestartOnFailure,
			Backoff: component.DialTaskBackoffConfig,
		}
	}
	if a.forwarderConfig.Enable && a.dataPlaneAddress != "" {
		c.RegisterTask(newTaskConfig("pb_publish_uplink", a.publishUplink))
		c.RegisterTask(newTaskConfig("pb_subscribe_downlink", a.subscribeDownlink))
	}
	if a.homeNetworkConfig.Enable && a.dataPlaneAddress != "" {
		c.RegisterTask(newTaskConfig("pb_subscribe_uplink", a.subscribeUplink))
		c.RegisterTask(newTaskConfig("pb_publish_downlink", a.publishDownlink))
	}
	hooks.RegisterUnaryHook("/ttn.lorawan.v3.GsPba", cluster.HookName, c.ClusterAuthUnaryHook())
	hooks.RegisterUnaryHook("/ttn.lorawan.v3.NsPba", cluster.HookName, c.ClusterAuthUnaryHook())
	c.RegisterGRPC(a)
	return a, nil
}
func (a *Agent) Context() context.Context {
	return a.ctx
}
func (a *Agent) Roles() []ttnpb.ClusterRole {
	return []ttnpb.ClusterRole{ttnpb.ClusterRole_PACKET_BROKER_AGENT}
}
func (a *Agent) RegisterServices(s *grpc.Server) {
	ttnpb.RegisterPbaServer(s, a.grpc.pba)
	ttnpb.RegisterNsPbaServer(s, a.grpc.nsPba)
	ttnpb.RegisterGsPbaServer(s, a.grpc.gsPba)
}