func (c *Connection) LocationChanged() <-chan struct{} {
	return c.locCh
}
LocationChanged returns the location updates channel.
package io
import (
	"bytes"
	"context"
	"encoding/base64"
	"fmt"
	"sync/atomic"
	"time"
	"github.com/mohae/deepcopy"
	"go.thethings.network/lorawan-stack/v3/pkg/band"
	"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/frequencyplans"
	"go.thethings.network/lorawan-stack/v3/pkg/gatewayserver/scheduling"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
	"go.thethings.network/lorawan-stack/v3/pkg/ratelimit"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
)
const (
	bufferSize = 1 << 4
	maxRTTs = 20
	rttTTL  = 30 * time.Minute
)
type Frontend interface {
	Protocol() string
	SupportsDownlinkClaim() bool
}
type Server interface {
	GetBaseConfig(ctx context.Context) config.ServiceBase
	FillGatewayContext(ctx context.Context, ids ttnpb.GatewayIdentifiers) (context.Context, ttnpb.GatewayIdentifiers, error)
	Connect(ctx context.Context, frontend Frontend, ids ttnpb.GatewayIdentifiers) (*Connection, error)
	GetFrequencyPlans(ctx context.Context, ids ttnpb.GatewayIdentifiers) (map[string]*frequencyplans.FrequencyPlan, error)
	ClaimDownlink(ctx context.Context, ids ttnpb.GatewayIdentifiers) error
	UnclaimDownlink(ctx context.Context, ids ttnpb.GatewayIdentifiers) error
	FromRequestContext(ctx context.Context) context.Context
	RateLimiter() ratelimit.Interface
	ValidateGatewayID(ctx context.Context, ids ttnpb.GatewayIdentifiers) error
	component.TaskStarter
}
type Connection struct {
	uplinks,
	downlinks uint64
	connectTime,
	lastStatusTime,
	lastUplinkTime,
	lastDownlinkTime int64
	lastStatus atomic.Value
	ctx       context.Context
	cancelCtx errorcontext.CancelFunc
	frontend         Frontend
	gateway          *ttnpb.Gateway
	gatewayPrimaryFP *frequencyplans.FrequencyPlan
	gatewayFPs       map[string]*frequencyplans.FrequencyPlan
	bandID           string
	fps              *frequencyplans.Store
	scheduler        *scheduling.Scheduler
	rtts             *rtts
	upCh     chan *ttnpb.GatewayUplinkMessage
	downCh   chan *ttnpb.DownlinkMessage
	statusCh chan *ttnpb.GatewayStatus
	txAckCh  chan *ttnpb.TxAcknowledgment
	statsChangedCh chan struct{}
	locCh          chan struct{}
	lastUplink            *uplinkMessage
	lastRepeatUpEventTime time.Time
}
type uplinkMessage struct {
	payload   []byte
	frequency uint64
	antennas  []uint32
}
var (
	errInconsistentFrequencyPlans = errors.DefineCorruption(
		"inconsistent_frequency_plans",
		"inconsistent frequency plans configuration",
	)
	errFrequencyPlansNotFromSameBand = errors.DefineInvalidArgument(
		"frequency_plans_not_from_same_band",
		"frequency plans must be from the same band",
	)
)
func NewConnection(ctx context.Context, frontend Frontend, gateway *ttnpb.Gateway, fps *frequencyplans.Store, enforceDutyCycle bool, scheduleAnytimeDelay *time.Duration) (*Connection, error) {
	gatewayFPs := make(map[string]*frequencyplans.FrequencyPlan, len(gateway.FrequencyPlanIds))
	fp0ID := gateway.FrequencyPlanId
	fp0, err := fps.GetByID(fp0ID)
	if err != nil {
		return nil, err
	}
	gatewayFPs[fp0ID] = fp0
	bandID := fp0.BandID
	if len(gateway.FrequencyPlanIds) > 0 {
		if gateway.FrequencyPlanIds[0] != fp0ID {
			return nil, errInconsistentFrequencyPlans.New()
		}
		for i := 1; i < len(gateway.FrequencyPlanIds); i++ {
			fpn, err := fps.GetByID(gateway.FrequencyPlanIds[i])
			if err != nil {
				return nil, err
			}
			if fpn.BandID != fp0.BandID {
				return nil, errFrequencyPlansNotFromSameBand.New()
			}
			gatewayFPs[gateway.FrequencyPlanIds[i]] = fpn
		}
	}
	ctx, cancelCtx := errorcontext.New(ctx)
	scheduler, err := scheduling.NewScheduler(ctx, gatewayFPs, enforceDutyCycle, scheduleAnytimeDelay, nil)
	if err != nil {
		return nil, err
	}
	return &Connection{
		ctx:       ctx,
		cancelCtx: cancelCtx,
		frontend:         frontend,
		gateway:          gateway,
		gatewayPrimaryFP: fp0,
		gatewayFPs:       gatewayFPs,
		bandID:           bandID,
		fps:              fps,
		scheduler:        scheduler,
		rtts:             newRTTs(maxRTTs, rttTTL),
		upCh:             make(chan *ttnpb.GatewayUplinkMessage, bufferSize),
		downCh:           make(chan *ttnpb.DownlinkMessage, bufferSize),
		statusCh:         make(chan *ttnpb.GatewayStatus, bufferSize),
		txAckCh:          make(chan *ttnpb.TxAcknowledgment, bufferSize),
		locCh:            make(chan struct{}, 1),
		connectTime:      time.Now().UnixNano(),
		statsChangedCh: make(chan struct{}, 1),
	}, nil
}
func (c *Connection) Context() context.Context { return c.ctx }
func (c *Connection) Disconnect(err error) {
	c.cancelCtx(err)
}
func (c *Connection) Frontend() Frontend { return c.frontend }
func (c *Connection) Gateway() *ttnpb.Gateway { return c.gateway }
var errBufferFull = errors.DefineInternal("buffer_full", "buffer is full")
const consecutiveRepeatUpEventsInterval = time.Minute
func (c *Connection) HandleUp(up *ttnpb.UplinkMessage) error {
	if c.discardRepeatedUplink(up) {
		return nil
	}
	var ct scheduling.ConcentratorTime
	if up.Settings.Time != nil {
		ct = c.scheduler.SyncWithGatewayAbsolute(up.Settings.Timestamp, up.ReceivedAt, *up.Settings.Time)
		log.FromContext(c.ctx).WithFields(log.Fields(
			"timestamp", up.Settings.Timestamp,
			"server_time", up.ReceivedAt,
			"gateway_time", *up.Settings.Time,
		)).Debug("Synchronized server and gateway absolute time")
	} else {
		ct = c.scheduler.Sync(up.Settings.Timestamp, up.ReceivedAt)
		log.FromContext(c.ctx).WithFields(log.Fields(
			"timestamp", up.Settings.Timestamp,
			"server_time", up.ReceivedAt,
		)).Debug("Synchronized server absolute time only")
	}
	for _, md := range up.RxMetadata {
		if md.AntennaIndex != 0 {
			md.DownlinkPathConstraint = ttnpb.DOWNLINK_PATH_CONSTRAINT_NEVER
			continue
		}
		buf, err := UplinkToken(ttnpb.GatewayAntennaIdentifiers{
			GatewayIdentifiers: *c.gateway.GetIds(),
			AntennaIndex:       md.AntennaIndex,
		}, md.Timestamp, ct, up.ReceivedAt)
		if err != nil {
			return err
		}
		md.UplinkToken = buf
		md.DownlinkPathConstraint = c.gateway.DownlinkPathConstraint
		if c.gateway.LocationPublic && len(c.gateway.Antennas) > int(md.AntennaIndex) {
			location := c.gateway.Antennas[md.AntennaIndex].Location
			if location != nil && location.Source != ttnpb.SOURCE_UNKNOWN {
				md.Location = location
			}
		} else if !c.gateway.LocationPublic {
			md.Location = nil
		}
	}
	msg := &ttnpb.GatewayUplinkMessage{
		UplinkMessage: up,
		BandId:        c.bandID,
	}
	select {
	case <-c.ctx.Done():
		return c.ctx.Err()
	case c.upCh <- msg:
		atomic.AddUint64(&c.uplinks, 1)
		atomic.StoreInt64(&c.lastUplinkTime, up.ReceivedAt.UnixNano())
		c.notifyStatsChanged()
	default:
		err := errBufferFull.New()
		registerDropMessage(c.ctx, c.gateway, "uplink", err)
		return err
	}
	return nil
}
func (c *Connection) HandleStatus(status *ttnpb.GatewayStatus) error {
	select {
	case <-c.ctx.Done():
		return c.ctx.Err()
	case c.statusCh <- status:
		c.lastStatus.Store(deepcopy.Copy(status))
		atomic.StoreInt64(&c.lastStatusTime, time.Now().UnixNano())
		c.notifyStatsChanged()
		if len(status.AntennaLocations) > 0 && c.gateway.UpdateLocationFromStatus {
			select {
			case c.locCh <- struct{}{}:
			default:
			}
		}
	default:
		err := errBufferFull.New()
		registerDropMessage(c.ctx, c.gateway, "status", err)
		return err
	}
	return nil
}
func (c *Connection) HandleTxAck(ack *ttnpb.TxAcknowledgment) error {
	select {
	case <-c.ctx.Done():
		return c.ctx.Err()
	case c.txAckCh <- ack:
		c.notifyStatsChanged()
	default:
		err := errBufferFull.New()
		registerDropMessage(c.ctx, c.gateway, "txack", err)
		return err
	}
	return nil
}
func (c *Connection) RecordRTT(d time.Duration, t time.Time) {
	c.rtts.Record(d, t)
	c.notifyStatsChanged()
}
var (
	errNotAllowed       = errors.DefineFailedPrecondition("not_allowed", "downlink not allowed")
	errNotTxRequest     = errors.DefineInvalidArgument("not_tx_request", "downlink message is not a Tx request")
	errNoAbsoluteTime   = errors.DefineInvalidArgument("no_absolute_time", "no absolute time provided for class B downlink")
	errNoGPSSync        = errors.DefineFailedPrecondition("no_gps_sync", "gateway time is not GPS synchronized")
	errNoRxDelay        = errors.DefineInvalidArgument("no_rx_delay", "no Rx delay provided for class A downlink")
	errNoUplinkToken    = errors.DefineInvalidArgument("no_uplink_token", "no uplink token provided for class A downlink")
	errDownlinkPath     = errors.DefineInvalidArgument("downlink_path", "invalid downlink path")
	errRxEmpty          = errors.DefineFailedPrecondition("rx_empty", "settings empty")
	errRxWindowSchedule = errors.Define("rx_window_schedule", "schedule in Rx window `{window}` failed")
	errDataRateRxWindow = errors.DefineInvalidArgument("data_rate_rx_window", "invalid data rate in Rx window `{window}`")
	errDataRateIndex    = errors.DefineInvalidArgument("data_rate_index", "no data rate with index `{index}`")
	errTooLong          = errors.DefineInvalidArgument("too_long", "the payload length `{payload_length}` exceeds maximum `{maximum_length}` at data rate index `{data_rate_index}`")
	errTxSchedule       = errors.DefineAborted("tx_schedule", "failed to schedule")
)
func getDownlinkPath(path *ttnpb.DownlinkPath, class ttnpb.Class) (ttnpb.GatewayAntennaIdentifiers, *ttnpb.UplinkToken, error) {
	if buf := path.GetUplinkToken(); len(buf) == 0 {
		if class == ttnpb.CLASS_A {
			return ttnpb.GatewayAntennaIdentifiers{}, nil, errNoUplinkToken.New()
		}
	} else {
		token, err := ParseUplinkToken(buf)
		if err != nil {
			return ttnpb.GatewayAntennaIdentifiers{}, nil, err
		}
		return token.GatewayAntennaIdentifiers, token, err
	}
	fixed := path.GetFixed()
	if fixed == nil {
		return ttnpb.GatewayAntennaIdentifiers{}, nil, errDownlinkPath.New()
	}
	return *fixed, nil, nil
}
func (c *Connection) SendDown(msg *ttnpb.DownlinkMessage) error {
	select {
	case <-c.ctx.Done():
		return c.ctx.Err()
	case c.downCh <- msg:
		atomic.AddUint64(&c.downlinks, 1)
		atomic.StoreInt64(&c.lastDownlinkTime, time.Now().UnixNano())
		c.notifyStatsChanged()
	default:
		return errBufferFull.New()
	}
	return nil
}
var (
	errFrequencyPlanNotConfigured   = errors.DefineInvalidArgument("frequency_plan_not_configured", "frequency plan `{id}` is not configured for this gateway")
	errNoFrequencyPlanIDInTxRequest = errors.DefineInvalidArgument("no_frequency_plan_id_in_tx_request", "no frequency plan ID in tx request")
)
func (c *Connection) ScheduleDown(path *ttnpb.DownlinkPath, msg *ttnpb.DownlinkMessage) (rx1, rx2 bool, delay time.Duration, err error) {
	if c.gateway.DownlinkPathConstraint == ttnpb.DOWNLINK_PATH_CONSTRAINT_NEVER {
		return false, false, 0, errNotAllowed.New()
	}
	request := msg.GetRequest()
	if request == nil {
		return false, false, 0, errNotTxRequest.New()
	}
	logger := log.FromContext(c.ctx).WithField("class", request.Class)
	logger.Debug("Attempt to schedule downlink on gateway")
	ids, uplinkToken, err := getDownlinkPath(path, request.Class)
	if err != nil {
		return false, false, 0, err
	}
	var fp *frequencyplans.FrequencyPlan
	fpID := request.GetFrequencyPlanId()
	if fpID != "" {
		fp = c.gatewayFPs[fpID]
		if fp == nil {
			fp, err = c.fps.GetByID(fpID)
			if err != nil {
				return false, false, 0, errFrequencyPlanNotConfigured.WithCause(err).WithAttributes("id", request.FrequencyPlanId)
			}
			if fp.BandID != c.bandID {
				return false, false, 0, errFrequencyPlansNotFromSameBand.New()
			}
		}
	} else {
		if len(c.gatewayFPs) != 1 {
			return false, false, 0, errNoFrequencyPlanIDInTxRequest.New()
		}
		for _, v := range c.gatewayFPs {
			fp = v
			break
		}
	}
	lwPhyVersion := request.GetLorawanPhyVersion()
	if lwPhyVersion == ttnpb.PHY_UNKNOWN {
		lwPhyVersion = ttnpb.RP001_V1_1_REV_B
	}
	phy, err := band.Get(fp.BandID, lwPhyVersion)
	if err != nil {
		return false, false, 0, err
	}
	var rxErrs []errors.ErrorDetails
	for i, rx := range []struct {
		dataRate      *ttnpb.DataRate
		dataRateIndex ttnpb.DataRateIndex
		frequency     uint64
		delay         time.Duration
	}{
		{
			dataRate: request.Rx1DataRate,
			dataRateIndex: request.Rx1DataRateIndex,
			frequency:     request.Rx1Frequency,
			delay:         0,
		},
		{
			dataRate: request.Rx2DataRate,
			dataRateIndex: request.Rx2DataRateIndex,
			frequency:     request.Rx2Frequency,
			delay:         time.Second,
		},
	} {
		if rx.frequency == 0 {
			rxErrs = append(rxErrs, errRxEmpty)
			continue
		}
		if rx.dataRate == nil {
			dr, ok := phy.DataRates[rx.dataRateIndex]
			if !ok {
				return false, false, 0, errDataRateIndex.WithAttributes("index", rx.dataRateIndex)
			}
			rx.dataRate = &dr.Rate
		}
		var (
			ok     bool
			bandDR band.DataRate
		)
		rx.dataRateIndex, bandDR, ok = phy.FindDownlinkDataRate(*rx.dataRate)
		if !ok {
			return false, false, 0, errDataRateRxWindow.WithAttributes("window", i+1)
		}
		logger := logger.WithFields(log.Fields(
			"rx_window", i+1,
			"frequency", rx.frequency,
			"data_rate_index", rx.dataRateIndex,
		))
		logger.Debug("Attempt to schedule downlink in receive window")
		maxPHYLength := bandDR.MaxMACPayloadSize(fp.DwellTime.GetDownlinks()) + 5
		if len(msg.RawPayload) > int(maxPHYLength) {
			return false, false, 0, errTooLong.WithAttributes(
				"payload_length", len(msg.RawPayload),
				"maximum_length", maxPHYLength,
				"data_rate_index", rx.dataRateIndex,
			)
		}
		eirp := phy.DefaultMaxEIRP
		if sb, ok := phy.FindSubBand(rx.frequency); ok {
			eirp = sb.MaxEIRP
		}
		if fp.MaxEIRP != nil {
			eirp = *fp.MaxEIRP
		}
		if sb, ok := fp.FindSubBand(rx.frequency); ok && sb.MaxEIRP != nil {
			eirp = *sb.MaxEIRP
		}
		settings := ttnpb.TxSettings{
			DataRate: *rx.dataRate,
			DataRateIndex: rx.dataRateIndex,
			Frequency:     rx.frequency,
			Downlink: &ttnpb.TxSettings_Downlink{
				TxPower:      eirp,
				AntennaIndex: ids.AntennaIndex,
			},
		}
		if int(ids.AntennaIndex) < len(c.gateway.Antennas) {
			settings.Downlink.TxPower -= c.gateway.Antennas[ids.AntennaIndex].Gain
		}
		switch mod := rx.dataRate.Modulation.(type) {
		case *ttnpb.DataRate_Lora:
			settings.CodingRate = phy.LoRaCodingRate
			settings.Downlink.InvertPolarization = true
		case *ttnpb.DataRate_Lrfhss:
			settings.CodingRate = mod.Lrfhss.CodingRate
		}
		var f func(context.Context, scheduling.Options) (scheduling.Emission, error)
		switch request.Class {
		case ttnpb.CLASS_A:
			f = c.scheduler.ScheduleAt
			if request.Rx1Delay == ttnpb.RX_DELAY_0 {
				return false, false, 0, errNoRxDelay.New()
			}
			settings.Timestamp = uplinkToken.Timestamp + uint32((time.Duration(request.Rx1Delay)*time.Second+rx.delay)/time.Microsecond)
		case ttnpb.CLASS_B:
			if request.AbsoluteTime == nil {
				return false, false, 0, errNoAbsoluteTime.New()
			}
			if !c.scheduler.IsGatewayTimeSynced() {
				rxErrs = append(rxErrs, errNoGPSSync)
				continue
			}
			f = c.scheduler.ScheduleAt
			settings.Time = request.AbsoluteTime
		case ttnpb.CLASS_C:
			if request.AbsoluteTime != nil {
				f = c.scheduler.ScheduleAt
				settings.Time = request.AbsoluteTime
			} else {
				f = c.scheduler.ScheduleAnytime
			}
		default:
			panic(fmt.Sprintf("proto: unexpected class %v in oneof", request.Class))
		}
		em, err := f(c.ctx, scheduling.Options{
			PayloadSize: len(msg.RawPayload),
			TxSettings:  settings,
			RTTs:        c.rtts,
			Priority:    request.Priority,
			UplinkToken: uplinkToken,  
		})
		if err != nil {
			logger.WithError(err).Debug("Failed to schedule downlink in Rx window")
			rxErrs = append(rxErrs, errRxWindowSchedule.WithCause(err).WithAttributes("window", i+1))
			continue
		}
		if settings.Time == nil || !c.scheduler.IsGatewayTimeSynced() {
			settings.Time = nil
			settings.Timestamp = uint32(time.Duration(em.Starts()) / time.Microsecond)
		} else {
			settings.Timestamp = 0
		}
		msg.Settings = &ttnpb.DownlinkMessage_Scheduled{
			Scheduled: &settings,
		}
		rx1 = i == 0
		rx2 = i == 1
		rxErrs = nil
		if now, ok := c.scheduler.Now(); ok {
			logger = logger.WithField("now", now)
			delay = time.Duration(em.Starts() - now)
		}
		logger.WithFields(log.Fields(
			"rx_window", i+1,
			"starts", em.Starts(),
			"duration", em.Duration(),
		)).Debug("Scheduled downlink")
		break
	}
	if len(rxErrs) > 0 {
		protoErrs := make([]*ttnpb.ErrorDetails, 0, len(rxErrs))
		for _, rxErr := range rxErrs {
			protoErrs = append(protoErrs, ttnpb.ErrorDetailsToProto(rxErr))
		}
		return false, false, 0, errTxSchedule.WithDetails(&ttnpb.ScheduleDownlinkErrorDetails{
			PathErrors: protoErrs,
		})
	}
	err = c.SendDown(msg)
	if err != nil {
		return false, false, 0, err
	}
	return
}
func (c *Connection) Status() <-chan *ttnpb.GatewayStatus {
	return c.statusCh
}
func (c *Connection) Up() <-chan *ttnpb.GatewayUplinkMessage {
	return c.upCh
}
func (c *Connection) Down() <-chan *ttnpb.DownlinkMessage {
	return c.downCh
}
func (c *Connection) TxAck() <-chan *ttnpb.TxAcknowledgment {
	return c.txAckCh
}
func (c *Connection) StatsChanged() <-chan struct{} {
	return c.statsChangedCh
}