pubsub_datasource

Documentation

Index

• type Configuration
• type Connector
• type EventConfiguration
• type EventType
  • func EventTypeFromString(s string) (EventType, error)
• type Factory
  • func NewFactory[T Configuration](executionContext context.Context, pubSubBySourceName map[string]PubSub) *Factory[T]
  • func (f *Factory[T]) Context() context.Context
  • func (f *Factory[T]) Planner(_ abstractlogger.Logger) plan.DataSourcePlanner[T]
• type Planner
  • func (p *Planner[T]) ConfigureFetch() resolve.FetchConfiguration
  • func (p *Planner[T]) ConfigureSubscription() plan.SubscriptionConfiguration
  • func (p *Planner[T]) DataSourcePlanningBehavior() plan.DataSourcePlanningBehavior
  • func (p *Planner[T]) DownstreamResponseFieldAlias(_ int) (alias string, exists bool)
  • func (p *Planner[T]) EnterDocument(_, _ *ast.Document)
  • func (p *Planner[T]) EnterField(ref int)
  • func (p *Planner[T]) Register(visitor *plan.Visitor, configuration plan.DataSourceConfiguration[T], ...) error
  • func (p *Planner[T]) UpstreamSchema(_ plan.DataSourceConfiguration[T]) (*ast.Document, bool)
• type PubSub
• type PublishAndRequestEventConfiguration
• type PublishDataSource
  • func (s *PublishDataSource) Load(ctx context.Context, input []byte, w io.Writer) error
• type RequestDataSource
  • func (s *RequestDataSource) Load(ctx context.Context, input []byte, w io.Writer) error
• type StreamConfiguration
• type SubscriptionEventConfiguration
• type SubscriptionSource
  • func (s *SubscriptionSource) Start(ctx *resolve.Context, input []byte, updater resolve.SubscriptionUpdater) error
  • func (s *SubscriptionSource) UniqueRequestID(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error
• type SubscriptionSourceInput

Types

type Configuration

type Configuration struct {
	Events []EventConfiguration `json:"events"`
}

type Connector

type Connector interface {
	New(ctx context.Context) PubSub
}

type EventConfiguration

type EventConfiguration struct {
	FieldName           string               `json:"fieldName"`
	SourceName          string               `json:"sourceName"`
	StreamConfiguration *StreamConfiguration `json:"streamConfiguration"`
	Subjects            []string             `json:"subjects"`
	Type                EventType            `json:"type"`
	TypeName            string               `json:"typeName"`
}

type EventType

type EventType string

const (
	EventTypePublish   EventType = "publish"
	EventTypeRequest   EventType = "request"
	EventTypeSubscribe EventType = "subscribe"
)

func EventTypeFromString

func EventTypeFromString(s string) (EventType, error)

type Factory

type Factory[T Configuration] struct {
	PubSubBySourceName map[string]PubSub
	// contains filtered or unexported fields
}

func NewFactory

func NewFactory[T Configuration](executionContext context.Context, pubSubBySourceName map[string]PubSub) *Factory[T]

func (*Factory[T]) Context

func (f *Factory[T]) Context() context.Context

func (*Factory[T]) Planner

func (f *Factory[T]) Planner(_ abstractlogger.Logger) plan.DataSourcePlanner[T]

type Planner

type Planner[T Configuration] struct {
	// contains filtered or unexported fields
}

func (*Planner[T]) ConfigureFetch

func (p *Planner[T]) ConfigureFetch() resolve.FetchConfiguration

func (*Planner[T]) ConfigureSubscription

func (p *Planner[T]) ConfigureSubscription() plan.SubscriptionConfiguration

func (*Planner[T]) DataSourcePlanningBehavior

func (p *Planner[T]) DataSourcePlanningBehavior() plan.DataSourcePlanningBehavior

func (*Planner[T]) DownstreamResponseFieldAlias

func (p *Planner[T]) DownstreamResponseFieldAlias(_ int) (alias string, exists bool)

func (*Planner[T]) EnterDocument

func (p *Planner[T]) EnterDocument(_, _ *ast.Document)

func (*Planner[T]) EnterField

func (p *Planner[T]) EnterField(ref int)

func (*Planner[T]) Register

func (p *Planner[T]) Register(visitor *plan.Visitor, configuration plan.DataSourceConfiguration[T], dataSourcePlannerConfiguration plan.DataSourcePlannerConfiguration) error

func (*Planner[T]) UpstreamSchema

func (p *Planner[T]) UpstreamSchema(_ plan.DataSourceConfiguration[T]) (*ast.Document, bool)

type PubSub

type PubSub interface {
	// ID is the unique identifier of the pubsub implementation (e.g. NATS)
	// This is used to uniquely identify a subscription
	ID() string
	// Subscribe starts listening on the given subjects and sends the received messages to the given next channel
	Subscribe(ctx context.Context, subjects []string, updater resolve.SubscriptionUpdater, streamConfiguration *StreamConfiguration) error
	// Publish sends the given data to the given subject
	Publish(ctx context.Context, subject string, data []byte) error
	// Request sends a request on the given subject and writes the response to the given writer
	Request(ctx context.Context, subject string, data []byte, w io.Writer) error
}

PubSub describe the interface that implements the primitive operations for pubsub

type PublishAndRequestEventConfiguration

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

type PublishDataSource

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

func (*PublishDataSource) Load

func (s *PublishDataSource) Load(ctx context.Context, input []byte, w io.Writer) error

type RequestDataSource

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

func (*RequestDataSource) Load

func (s *RequestDataSource) Load(ctx context.Context, input []byte, w io.Writer) error

type StreamConfiguration

type StreamConfiguration struct {
	Consumer   string
	StreamName string
}

type SubscriptionEventConfiguration

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

type SubscriptionSource

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

func (*SubscriptionSource) Start

func (s *SubscriptionSource) Start(ctx *resolve.Context, input []byte, updater resolve.SubscriptionUpdater) error

func (*SubscriptionSource) UniqueRequestID

func (s *SubscriptionSource) UniqueRequestID(ctx *resolve.Context, input []byte, xxh *xxhash.Digest) error

type SubscriptionSourceInput

type SubscriptionSourceInput struct {
	Subjects            []string             `json:"subjects"`
	SourceName          string               `json:"sourceName"`
	StreamConfiguration *StreamConfiguration `json:"streamConfiguration"`
}