io

Documentation

Index

• Constants
• Variables
• func CleanDownlinks(items []*ttnpb.ApplicationDownlink) []*ttnpb.ApplicationDownlink
• type Cluster
• type ContextualApplicationUp
• type DownlinkQueueOperator
• type EndDeviceRegistry
• type PubSub
• type Server
• type Subscription
  • func NewSubscription(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, ...) *Subscription
  • func (s *Subscription) ApplicationIDs() *ttnpb.ApplicationIdentifiers
  • func (s *Subscription) Context() context.Context
  • func (s *Subscription) Disconnect(err error)
  • func (s *Subscription) Pipe(ctx context.Context, ts task.Starter, name string, ...)
  • func (s *Subscription) Protocol() string
  • func (s *Subscription) Publish(ctx context.Context, up *ttnpb.ApplicationUp) error
  • func (s *Subscription) Up() <-chan *ContextualApplicationUp
• type SubscriptionOption
  • func WithBlocking(blocking bool) SubscriptionOption
  • func WithBufferSize(bufferSize int) SubscriptionOption
• type UplinkStorage

Constants

const DefaultBufferSize = 128

DefaultBufferSize is the default size of a subscription uplink buffer.

Variables

var (

	// DialTaskBackoffConfig derives the component.DialTaskBackoffConfig and dynamically determines the backoff duration
	// based on recent error codes.
	DialTaskBackoffConfig = &task.BackoffConfig{
		Jitter: task.DefaultBackoffJitter,
		IntervalFunc: func(ctx context.Context, executionDuration time.Duration, invocation uint, err error) time.Duration {
			intervals := dialTaskBackoffIntervals
			switch {
			case errors.IsFailedPrecondition(err),
				errors.IsUnauthenticated(err),
				errors.IsPermissionDenied(err),
				errors.IsInvalidArgument(err),
				errors.IsAlreadyExists(err),
				errors.IsCanceled(err),
				errors.IsNotFound(err):
				intervals = dialTaskExtendedBackoffIntervals
			}
			switch {
			case executionDuration > task.DefaultBackoffResetDuration:
				return intervals[0]
			case invocation >= uint(len(intervals)):
				return intervals[len(intervals)-1]
			default:
				return intervals[invocation-1]
			}
		},
	}
)

Functions

func CleanDownlinks

func CleanDownlinks(items []*ttnpb.ApplicationDownlink) []*ttnpb.ApplicationDownlink

CleanDownlinks returns a copy of the given downlink items with only the fields that can be set by the application.

Types

type Cluster

type Cluster interface {
	// GetPeers returns peers with the given role.
	GetPeers(ctx context.Context, role ttnpb.ClusterRole) ([]cluster.Peer, error)
	// GetPeer returns a peer with the given role, and a responsibility for the
	// given identifiers. If the identifiers are nil, this function returns a random
	// peer from the list that would be returned by GetPeers.
	GetPeer(ctx context.Context, role ttnpb.ClusterRole, ids cluster.EntityIdentifiers) (cluster.Peer, error)
	// GetPeerConn returns the gRPC client connection of a peer, if the peer is available as
	// as per GetPeer.
	GetPeerConn(ctx context.Context, role ttnpb.ClusterRole, ids cluster.EntityIdentifiers) (*grpc.ClientConn, error)
}

Cluster represents the Application Server cluster peers to application frontends.

type ContextualApplicationUp

type ContextualApplicationUp struct {
	context.Context
	*ttnpb.ApplicationUp
}

ContextualApplicationUp represents an ttnpb.ApplicationUp with its context.

type DownlinkQueueOperator

type DownlinkQueueOperator interface {
	// DownlinkQueuePush pushes the given downlink messages to the end device's application downlink queue.
	DownlinkQueuePush(context.Context, *ttnpb.EndDeviceIdentifiers, []*ttnpb.ApplicationDownlink) error
	// DownlinkQueueReplace replaces the end device's application downlink queue with the given downlink messages.
	DownlinkQueueReplace(context.Context, *ttnpb.EndDeviceIdentifiers, []*ttnpb.ApplicationDownlink) error
	// DownlinkQueueList lists the application downlink queue of the given end device.
	DownlinkQueueList(context.Context, *ttnpb.EndDeviceIdentifiers) ([]*ttnpb.ApplicationDownlink, error)
}

DownlinkQueueOperator represents the Application Server downlink queue operations to application frontends.

type EndDeviceRegistry

type EndDeviceRegistry interface {
	// GetEndDevice retrieves the end device from the Application Server end device registry.
	// This call will be delegated to the underlying end device registry, and should not be
	// used on the hot path. It exists for provisioning purposes.
	GetEndDevice(ctx context.Context, ids *ttnpb.EndDeviceIdentifiers, paths []string) (*ttnpb.EndDevice, error)
}

EndDeviceRegistry represents the Application Server end device registry to application frontends.

type PubSub

type PubSub interface {
	// Publish publishes upstream traffic to the Application Server.
	Publish(ctx context.Context, up *ttnpb.ApplicationUp) error
	// Subscribe subscribes an application or integration by its identifiers to the Application Server, and returns a
	// Subscription for traffic and control. If the cluster parameter is true, the subscription receives all of the
	// traffic of the application. Otherwise, only traffic that was processed locally is sent.
	Subscribe(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, cluster bool) (*Subscription, error)
}

PubSub represents the Application Server Pub/Sub capabilities to application frontends.

type Server

type Server interface {
	task.Starter
	httpclient.Provider
	PubSub
	DownlinkQueueOperator
	UplinkStorage
	Cluster
	EndDeviceRegistry
	// FromRequestContext decouples the lifetime of the provided context from the values found in the context.
	FromRequestContext(context.Context) context.Context
	// GetBaseConfig returns the component configuration.
	GetBaseConfig(ctx context.Context) config.ServiceBase
	// FillContext fills the given context.
	// This method should only be used for request contexts.
	FillContext(ctx context.Context) context.Context
	// RateLimiter returns the rate limiter instance.
	RateLimiter() ratelimit.Interface
}

Server represents the Application Server to application frontends.

type Subscription

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription is a subscription to an application or integration managed by a frontend.

func NewSubscription

func NewSubscription(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, opts ...SubscriptionOption) *Subscription

NewSubscription instantiates a new application or integration subscription.

func (*Subscription) ApplicationIDs

func (s *Subscription) ApplicationIDs() *ttnpb.ApplicationIdentifiers

ApplicationIDs returns the application identifiers, if the subscription represents any specific.

func (*Subscription) Context

func (s *Subscription) Context() context.Context

Context returns the subscription context.

func (*Subscription) Disconnect

func (s *Subscription) Disconnect(err error)

Disconnect marks the subscription as disconnected and cancels the context.

func (*Subscription) Pipe

func (s *Subscription) Pipe(ctx context.Context, ts task.Starter, name string, submit func(context.Context, interface{}) error)

Pipe pipes the output of the Subscription to the provided handler.

func (*Subscription) Protocol

func (s *Subscription) Protocol() string

Protocol returns the protocol used for the subscription, i.e. grpc, mqtt or http.

func (*Subscription) Publish

func (s *Subscription) Publish(ctx context.Context, up *ttnpb.ApplicationUp) error

Publish publishes an upstream message.

func (*Subscription) Up

func (s *Subscription) Up() <-chan *ContextualApplicationUp

Up returns the upstream channel.

type SubscriptionOption

type SubscriptionOption interface {
	// contains filtered or unexported methods
}

SubscriptionOption is an option for a Subscription.

func WithBlocking

func WithBlocking(blocking bool) SubscriptionOption

WithBlocking controls if the Publish call is blocking or not.

func WithBufferSize

func WithBufferSize(bufferSize int) SubscriptionOption

WithBufferSize controls the size of the subscription buffer.

type UplinkStorage

type UplinkStorage interface {
	// RangeUplinks ranges the application uplinks and calls the callback function, until false is returned.
	RangeUplinks(ctx context.Context, ids *ttnpb.EndDeviceIdentifiers, paths []string, f func(ctx context.Context, up *ttnpb.ApplicationUplink) bool) error
}

UplinkStorage represents the Application Server uplink storage to application frontends.