subscription

Documentation

Index

• Constants
• type Client
• type Executor
• type ExecutorPool
• type ExecutorV1
  • func (e *ExecutorV1) Execute(writer resolve.FlushWriter) error
  • func (e *ExecutorV1) OperationType() ast.OperationType
  • func (e *ExecutorV1) Reset()
  • func (e *ExecutorV1) SetContext(context context.Context)
• type ExecutorV1Pool
  • func NewExecutorV1Pool(executionHandler *execution.Handler) *ExecutorV1Pool
  • func (e *ExecutorV1Pool) Get(payload []byte) (Executor, error)
  • func (e *ExecutorV1Pool) Put(executor Executor) error
• type ExecutorV2
  • func (e *ExecutorV2) Execute(writer resolve.FlushWriter) error
  • func (e *ExecutorV2) OperationType() ast.OperationType
  • func (e *ExecutorV2) Reset()
  • func (e *ExecutorV2) SetContext(context context.Context)
• type ExecutorV2Pool
  • func NewExecutorV2Pool(engine *graphql.ExecutionEngineV2, connectionInitReqCtx context.Context) *ExecutorV2Pool
  • func (e *ExecutorV2Pool) Get(payload []byte) (Executor, error)
  • func (e *ExecutorV2Pool) Put(executor Executor) error
• type Handler
  • func NewHandler(logger abstractlogger.Logger, client Client, executorPool ExecutorPool) (*Handler, error)
  • func NewHandlerWithInitFunc(logger abstractlogger.Logger, client Client, executorPool ExecutorPool, ...) (*Handler, error)
  • func (h *Handler) ActiveSubscriptions() int
  • func (h *Handler) ChangeKeepAliveInterval(d time.Duration)
  • func (h *Handler) ChangeSubscriptionUpdateInterval(d time.Duration)
  • func (h *Handler) Handle(ctx context.Context)
• type InitPayload
  • func (p InitPayload) Authorization() string
  • func (p InitPayload) GetString(key string) string
• type InitialHttpRequestContext
  • func NewInitialHttpRequestContext(r *http.Request) *InitialHttpRequestContext
• type Message
• type WebsocketInitFunc

Constants

const (
	MessageTypeConnectionInit      = "connection_init"
	MessageTypeConnectionAck       = "connection_ack"
	MessageTypeConnectionError     = "connection_error"
	MessageTypeConnectionTerminate = "connection_terminate"
	MessageTypeConnectionKeepAlive = "ka"
	MessageTypeStart               = "start"
	MessageTypeStop                = "stop"
	MessageTypeData                = "data"
	MessageTypeError               = "error"
	MessageTypeComplete            = "complete"

	DefaultKeepAliveInterval          = "15s"
	DefaultSubscriptionUpdateInterval = "1s"
)

Types

type Client

type Client interface {
	// ReadFromClient will invoke a read operation from the client connection.
	ReadFromClient() (*Message, error)
	// WriteToClient will invoke a write operation to the client connection.
	WriteToClient(Message) error
	// IsConnected will indicate if a connection is still established.
	IsConnected() bool
	// Disconnect will close the connection between server and client.
	Disconnect() error
}

client provides an interface which can be implemented by any possible subscription client like websockets, mqtt, etc.

type Executor

type Executor interface {
	Execute(writer resolve.FlushWriter) error
	OperationType() ast.OperationType
	SetContext(context context.Context)
	Reset()
}

Executor is an abstraction for executing a GraphQL engine

type ExecutorPool

type ExecutorPool interface {
	Get(payload []byte) (Executor, error)
	Put(executor Executor) error
}

ExecutorPool is an abstraction for creating executors

type ExecutorV1

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

func (*ExecutorV1) Execute

func (e *ExecutorV1) Execute(writer resolve.FlushWriter) error

func (*ExecutorV1) OperationType

func (e *ExecutorV1) OperationType() ast.OperationType

func (*ExecutorV1) Reset

func (e *ExecutorV1) Reset()

func (*ExecutorV1) SetContext

func (e *ExecutorV1) SetContext(context context.Context)

type ExecutorV1Pool

type ExecutorV1Pool struct {
	ExecutionHandler *execution.Handler
	// contains filtered or unexported fields
}

func NewExecutorV1Pool

func NewExecutorV1Pool(executionHandler *execution.Handler) *ExecutorV1Pool

func (*ExecutorV1Pool) Get

func (e *ExecutorV1Pool) Get(payload []byte) (Executor, error)

func (*ExecutorV1Pool) Put

func (e *ExecutorV1Pool) Put(executor Executor) error

type ExecutorV2

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

func (*ExecutorV2) Execute

func (e *ExecutorV2) Execute(writer resolve.FlushWriter) error

func (*ExecutorV2) OperationType

func (e *ExecutorV2) OperationType() ast.OperationType

func (*ExecutorV2) Reset

func (e *ExecutorV2) Reset()

func (*ExecutorV2) SetContext

func (e *ExecutorV2) SetContext(context context.Context)

type ExecutorV2Pool

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

ExecutorV2Pool - provides reusable executors

func NewExecutorV2Pool

func NewExecutorV2Pool(engine *graphql.ExecutionEngineV2, connectionInitReqCtx context.Context) *ExecutorV2Pool

func (*ExecutorV2Pool) Get

func (e *ExecutorV2Pool) Get(payload []byte) (Executor, error)

func (*ExecutorV2Pool) Put

func (e *ExecutorV2Pool) Put(executor Executor) error

type Handler

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

Handler is the actual subscription handler which will keep track on how to handle messages coming from the client.

func NewHandler

func NewHandler(logger abstractlogger.Logger, client Client, executorPool ExecutorPool) (*Handler, error)

NewHandler creates a new subscription handler.

func NewHandlerWithInitFunc

func NewHandlerWithInitFunc(
	logger abstractlogger.Logger,
	client Client,
	executorPool ExecutorPool,
	initFunc WebsocketInitFunc,
) (*Handler, error)

func (*Handler) ActiveSubscriptions

func (h *Handler) ActiveSubscriptions() int

ActiveSubscriptions will return the actual number of active subscriptions for that client.

func (*Handler) ChangeKeepAliveInterval

func (h *Handler) ChangeKeepAliveInterval(d time.Duration)

ChangeKeepAliveInterval can be used to change the keep alive interval.

func (*Handler) ChangeSubscriptionUpdateInterval

func (h *Handler) ChangeSubscriptionUpdateInterval(d time.Duration)

ChangeSubscriptionUpdateInterval can be used to change the update interval.

func (*Handler) Handle

func (h *Handler) Handle(ctx context.Context)

Handle will handle the subscription connection.

type InitPayload

type InitPayload json.RawMessage

InitPayload is a structure that is parsed from the websocket init message payload.

func (InitPayload) Authorization

func (p InitPayload) Authorization() string

Authorization is a short hand for getting the Authorization header from the payload.

func (InitPayload) GetString

func (p InitPayload) GetString(key string) string

GetString safely gets a string value from the payload. It returns an empty string if the payload is nil or the value isn't set.

type InitialHttpRequestContext

type InitialHttpRequestContext struct {
	context.Context
	Request *http.Request
}

func NewInitialHttpRequestContext

func NewInitialHttpRequestContext(r *http.Request) *InitialHttpRequestContext

type Message

type Message struct {
	Id      string          `json:"id"`
	Type    string          `json:"type"`
	Payload json.RawMessage `json:"payload"`
}

Message defines the actual subscription message wich will be passed from client to server and vice versa.

type WebsocketInitFunc

type WebsocketInitFunc func(ctx context.Context, initPayload InitPayload) (context.Context, error)

WebsocketInitFunc is called when the server receives connection init message from the client. This can be used to check initial payload to see whether to accept the websocket connection.