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

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
	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, *ttnpb.ApplicationUp) 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.