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
}
HandleUp updates the uplink stats and sends the message to the upstream 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