stream

Documentation

Index

• Constants
• func DiscardReplies(any, *streamv1.BroadcastReplyList) error
• func NewAgentPlugin(p StreamAPIExtension) plugin.Plugin
• func NewGatewayPlugin(p StreamAPIExtension, opts ...GatewayStreamApiExtensionPluginOption) plugin.Plugin
• type Aggregator
• type GatewayStreamApiExtensionPluginOption
  • func WithMetrics(conf GatewayStreamMetricsConfig) GatewayStreamApiExtensionPluginOption
• type GatewayStreamApiExtensionPluginOptions
• type GatewayStreamMetricsConfig
• type Server
• type StreamAPIExtension
• type StreamAPIExtensionWithHandlers
• type StreamClientHandler
• type StreamDelegate
  • func NewDelegate[T any](cc grpc.ClientConnInterface, newClientFunc func(grpc.ClientConnInterface) T) StreamDelegate[T]

Constants

const (
	StreamAPIExtensionPluginID = "opni.apiextensions.StreamAPIExtension"
	ServiceID                  = "apiextensions.StreamAPIExtension"
)

const (
	CorrelationIDHeader = "x-correlation-id"
)

Functions

func DiscardReplies

func DiscardReplies(any, *streamv1.BroadcastReplyList) error

func NewAgentPlugin

func NewAgentPlugin(p StreamAPIExtension) plugin.Plugin

func NewGatewayPlugin

func NewGatewayPlugin(p StreamAPIExtension, opts ...GatewayStreamApiExtensionPluginOption) plugin.Plugin

Types

type Aggregator

type Aggregator func(reply any, msg *streamv1.BroadcastReplyList) error

Aggregator allows the user to aggregate the responses from a broadcast request, and store the result in reply.

type GatewayStreamApiExtensionPluginOption

type GatewayStreamApiExtensionPluginOption func(*GatewayStreamApiExtensionPluginOptions)

func WithMetrics

func WithMetrics(conf GatewayStreamMetricsConfig) GatewayStreamApiExtensionPluginOption

type GatewayStreamApiExtensionPluginOptions

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

type GatewayStreamMetricsConfig

type GatewayStreamMetricsConfig struct {
	// Prometheus registerer
	Reader metric.Reader

	// A function called on each stream's Connect that returns a list of static
	// labels to attach to all metrics collected for that stream.
	LabelsForStream func(context.Context) []attribute.KeyValue
}

type Server

type Server struct {
	Desc              *grpc.ServiceDesc
	Impl              interface{}
	RequireCapability string
}

type StreamAPIExtension

type StreamAPIExtension interface {
	StreamServers() []Server
}

type StreamAPIExtensionWithHandlers

type StreamAPIExtensionWithHandlers interface {
	StreamAPIExtension
	StreamClientHandler
}

type StreamClientHandler

type StreamClientHandler interface {
	UseStreamClient(client grpc.ClientConnInterface)
}

A plugin can optionally implement StreamClientHandler to obtain a grpc.ClientConnInterface to the plugin's side of the spliced stream.

type StreamDelegate

type StreamDelegate[T any] interface {
	WithTarget(*corev1.Reference) T
	WithBroadcastSelector(*corev1.ClusterSelector, Aggregator) T
}

func NewDelegate

func NewDelegate[T any](cc grpc.ClientConnInterface, newClientFunc func(grpc.ClientConnInterface) T) StreamDelegate[T]