wsclient

package
v0.0.0-...-0f876b5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 23, 2025 License: Apache-2.0 Imports: 26 Imported by: 3

Documentation

Overview

Package wsclient wraps the generated aws-sdk-go client to provide marshalling and unmarshalling of data over a websocket connection in the format expected by backend. It allows for bidirectional communication and acts as both a client-and-server in terms of requests, but only as a client in terms of connecting.

Package wsclient wraps the generated aws-sdk-go client to provide marshalling and unmarshalling of data over a websocket connection in the format expected by backend. It allows for bidirectional communication and acts as both a client-and-server in terms of requests, but only as a client in terms of connecting.

Index

Constants

View Source 
const (
	// ServiceName defines the service name for the agent. This is used to sign messages
	// that are sent to the backend.
	ServiceName = "ecs"

	// ExitTerminal indicates the agent run into error that's not recoverable
	// no need to restart
	ExitTerminal = 5

	// disconnectTimeout is the maximum time taken by the server side (TACS/ACS) to send a
	// disconnect payload for the Agent.
	DisconnectTimeout = 30 * time.Minute

	// disconnectJitterMax is the maximum jitter time chosen as reasonable initial value
	// to prevent mass retries at the same time from multiple clients/tasks synchronizing.
	DisconnectJitterMax = 5 * time.Minute
)

Variables

This section is empty.

Functions

func DecodeConnectionError 

func DecodeConnectionError(data []byte, dec TypeDecoder) (interface{}, string, error)

DecodeConnectionError decodes some of the connection errors returned by the backend. Some differ from the usual ones in that they do not have a 'type' and 'message' field, but rather are of the form {"ErrorType":"ErrorMessage"}

func DecodeData 

func DecodeData(data []byte, dec TypeDecoder) (interface{}, string, error)

DecodeData decodes a raw message into its type. E.g. An ACS message of the form {"type":"FooMessage","message":{"foo":1}} will be decoded into the corresponding *ecsacs.FooMessage type. The type string, "FooMessage", will also be returned as a convenience.

Types

type ClientFactory 

type ClientFactory interface {
	New(url string, credentialProvider *credentials.Credentials, rwTimeout time.Duration, cfg *WSClientMinAgentConfig, metricsFactory metrics.EntryFactory) ClientServer
}

ClientFactory interface abstracts the method that creates new ClientServer objects. This is helpful when writing unit tests.

type ClientServer 

type ClientServer interface {
	AddRequestHandler(RequestHandler)
	// SetAnyRequestHandler takes a function with the signature 'func(i
	// interface{})' and calls it with every message the server passes down.
	// Only a single 'AnyRequestHandler' will be active at a given time for a
	// ClientServer
	SetAnyRequestHandler(RequestHandler)
	MakeRequest(input interface{}) error
	WriteMessage(input []byte) error
	WriteCloseMessage() error
	Connect(disconnectMetricName string, disconnectTimeout time.Duration, disconnectJitterMax time.Duration) (*time.Timer, error)
	IsConnected() bool
	SetConnection(conn wsconn.WebsocketConn)
	Disconnect(...interface{}) error
	Serve(ctx context.Context) error
	SetReadDeadline(t time.Time) error
	CloseClient(t time.Time, dur time.Duration) error
	io.Closer
}

ClientServer is a combined client and server for the backend websocket connection

type ClientServerImpl 

type ClientServerImpl struct {
	// Cfg is the subset of user-specified runtime configuration
	Cfg *WSClientMinAgentConfig

	// CredentialProvider is used to retrieve AWS credentials
	CredentialProvider *credentials.Credentials
	// RequestHandlers is a map from message types to handler functions of the
	// form:
	//     "FooMessage": func(message *ecsacs.FooMessage)
	RequestHandlers map[string]RequestHandler
	// AnyRequestHandler is a request handler that, if set, is called on every
	// message with said message. It will be called before a RequestHandler is
	// called. It must take a single interface{} argument.
	AnyRequestHandler RequestHandler
	// MakeRequestHook is an optional callback that, if set, is called on every
	// generated request with the raw request body.
	MakeRequestHook MakeRequestHookFunc
	// URL is the full url to the backend, including path, querystring, and so on.
	URL string
	// RWTimeout is the duration used for setting read and write deadlines
	// for the websocket connection
	RWTimeout time.Duration

	// MetricsFactory needed to emit metrics for monitoring.
	MetricsFactory metrics.EntryFactory
	ClientServer
	ServiceError
	TypeDecoder
	// contains filtered or unexported fields
}

ClientServerImpl wraps commonly used methods defined in ClientServer interface.

func (*ClientServerImpl) AddRequestHandler 

func (cs *ClientServerImpl) AddRequestHandler(f RequestHandler)

AddRequestHandler adds a request handler to this client. A request handler *must* be a function taking a single argument, and that argument *must* be a pointer to a recognized 'ecsacs' struct. E.g. if you desired to handle messages from acs of type 'FooMessage', you would pass the following handler in: 

func(message *ecsacs.FooMessage)

This function will cause agent exit if the passed in function does not have one pointer argument or the argument is not a recognized type. Additionally, the request handler will block processing of further messages on this connection so it's important that it return quickly.

func (*ClientServerImpl) CloseClient 

func (cs *ClientServerImpl) CloseClient(startTime time.Time, timeoutDuration time.Duration) error

closeClient will attempt to close the provided client, retries are not recommended as failure modes for this are when client is not found or already closed.

func (*ClientServerImpl) Connect 

func (cs *ClientServerImpl) Connect(disconnectMetricName string,
	disconnectTimeout time.Duration,
	disconnectJitterMax time.Duration) (*time.Timer, error)

Connect opens a connection to the backend and upgrades it to a websocket. Calls to 'MakeRequest' can be made after calling this, but responses will not be receivable until 'Serve' is also called.

func (*ClientServerImpl) ConsumeMessages 

func (cs *ClientServerImpl) ConsumeMessages(ctx context.Context) error

ConsumeMessages reads messages from the websocket connection and handles read messages from an active connection.

func (*ClientServerImpl) CreateRequestMessage 

func (cs *ClientServerImpl) CreateRequestMessage(input interface{}) ([]byte, error)

CreateRequestMessage creates the request json message using the given input. Note, the input *MUST* be a pointer to a valid backend type that this client recognises.

func (*ClientServerImpl) Disconnect 

func (cs *ClientServerImpl) Disconnect(...interface{}) error

Disconnect disconnects the connection

func (*ClientServerImpl) IsReady 

func (cs *ClientServerImpl) IsReady() bool

IsReady gives a boolean response that informs the caller if the websocket connection is fully established.

func (*ClientServerImpl) MakeRequest 

func (cs *ClientServerImpl) MakeRequest(input interface{}) error

MakeRequest makes a request using the given input. Note, the input *MUST* be a pointer to a valid backend type that this client recognises

func (*ClientServerImpl) SetAnyRequestHandler 

func (cs *ClientServerImpl) SetAnyRequestHandler(f RequestHandler)

SetAnyRequestHandler passes a RequestHandler object into the client.

func (*ClientServerImpl) SetConnection 

func (cs *ClientServerImpl) SetConnection(conn wsconn.WebsocketConn)

SetConnection passes a websocket connection object into the client. This is used only in testing and should be avoided in non-test code.

func (*ClientServerImpl) SetReadDeadline 

func (cs *ClientServerImpl) SetReadDeadline(t time.Time) error

SetReadDeadline sets the read deadline for the websocket connection A read timeout results in an io error if there are any outstanding reads that exceed the deadline

func (*ClientServerImpl) WriteCloseMessage 

func (cs *ClientServerImpl) WriteCloseMessage() error

WriteCloseMessage wraps the low level websocket WriteControl method with a lock, and sends a message of type CloseMessage (Ref: https://github.com/gorilla/websocket/blob/9111bb834a68b893cebbbaed5060bdbc1d9ab7d2/conn.go#L74)

func (*ClientServerImpl) WriteMessage 

func (cs *ClientServerImpl) WriteMessage(send []byte) error

WriteMessage wraps the low level websocket write method with a lock

type MakeRequestHookFunc 

type MakeRequestHookFunc func([]byte) ([]byte, error)

MakeRequestHookFunc is a function that is invoked on every generated request with the raw request body. MakeRequestHookFunc must return either the body to send or an error.

type NotMarshallableWSRequest 

type NotMarshallableWSRequest struct {
	Type string

	Err error
}

NotMarshallableWSRequest represents that the given request input could not be marshalled

func (*NotMarshallableWSRequest) Error 

func (u *NotMarshallableWSRequest) Error() string

Error implements error

func (*NotMarshallableWSRequest) Retry 

func (u *NotMarshallableWSRequest) Retry() bool

Retry implements Retriable

type ReceivedMessage 

type ReceivedMessage struct {
	Type    string          `json:"type"`
	Message json.RawMessage `json:"message"`
}

ReceivedMessage is the intermediate message used to unmarshal a message from backend

type RequestHandler 

type RequestHandler interface{}

RequestHandler would be func(*ecsacs.T for T in ecsacs.*) to be more proper, but it needs to be interface{} to properly capture that

type RequestMessage 

type RequestMessage struct {
	Type    string          `json:"type"`
	Message json.RawMessage `json:"message"`
}

RequestMessage is the intermediate message marshalled to send to backend.

type RequestResponder 

type RequestResponder interface {
	// Name returns the name of the responder. This is used mostly for logging.
	Name() string
	// HandlerFunc returns the RequestHandler callback for a particular
	// websocket request message type.
	HandlerFunc() RequestHandler
}

RequestResponder wraps the RequestHandler interface with a Respond() method that can be used to Respond to requests read and processed via the RequestHandler interface for a particular message type.

Example: 

type payloadMessageDispatcher struct {
    respond  func(interface{}) error
    dispatcher actor.Dispatcher
}
func(d *payloadmessagedispatcher) HandlerFunc() RequestHandler {
    return func(payload *ecsacs.PayloadMessage) {
        message := &actor.DispatcherMessage{
            Payload: payload,
            AckFunc: func() error {
                return d.respond()
            },
            ...
        }
        d.dispatcher.Send(message)
    }
}

type RespondFunc 

type RespondFunc func(interface{}) error

RespondFunc specifies a function callback that can be used by the RequestResponder to respond to requests.

type ServiceError 

type ServiceError interface {
	NewError(err interface{}) *WSError
}

ServiceError defines methods to return new backend specific error objects.

type TypeDecoder 

type TypeDecoder interface {
	// NewOfType returns an object of a recognized type for a given type name.
	// It additionally returns a boolean value which is set to false for an
	// unrecognized type.
	NewOfType(string) (interface{}, bool)

	// GetRecognizedTypes returns a map of type-strings (as passed in acs/tcs messages as
	// the 'type' field) to a pointer to the corresponding struct type this type should
	// be marshalled/unmarshalled to/from.
	GetRecognizedTypes() map[string]reflect.Type
}

TypeDecoder interface defines methods to decode ecs types.

func BuildTypeDecoder 

func BuildTypeDecoder(recognizedTypes []interface{}) TypeDecoder

BuildTypeDecoder takes a list of interfaces and stores them internally as a list of typeMappings in the format below. "MyMessage": TypeOf(ecstcs.MyMessage)

type TypeDecoderImpl 

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

TypeDecoderImpl is an implementation for general use between ACS and TCS clients

func (*TypeDecoderImpl) GetRecognizedTypes 

func (d *TypeDecoderImpl) GetRecognizedTypes() map[string]reflect.Type

func (*TypeDecoderImpl) NewOfType 

func (d *TypeDecoderImpl) NewOfType(typeString string) (interface{}, bool)

type UndecodableMessage 

type UndecodableMessage struct {
	Msg string
}

UndecodableMessage indicates that a message from the backend could not be decoded

func (*UndecodableMessage) Error 

func (u *UndecodableMessage) Error() string

type UnrecognizedWSRequestType 

type UnrecognizedWSRequestType struct {
	Type string
}

UnrecognizedWSRequestType specifies that a given type is not recognized. This error is not retriable.

func (*UnrecognizedWSRequestType) Error 

func (u *UnrecognizedWSRequestType) Error() string

Error implements error

func (*UnrecognizedWSRequestType) Retry 

func (u *UnrecognizedWSRequestType) Retry() bool

Retry implements Retriable

type WSClientMinAgentConfig 

type WSClientMinAgentConfig struct {
	AWSRegion          string
	AcceptInsecureCert bool
	DockerEndpoint     string
	IsDocker           bool
}

WSClientMinAgentConfig is a subset of agent's config.

type WSError 

type WSError struct {
	ErrObj interface{}
	Type   string
	WSUnretriableErrors
}

WSError wraps all the typed errors that the backend may return This will not be needed once the aws-sdk-go generation handles error types more cleanly

func (*WSError) Error 

func (err *WSError) Error() string

Error returns an error string

func (*WSError) Retry 

func (err *WSError) Retry() bool

Retry returns true if this error should be considered retriable

type WSUnretriableErrors 

type WSUnretriableErrors interface {
	Get() []interface{}
}

WSUnretriableErrors defines methods to retrieve the list of unretriable errors.

Source Files

View all Source files

Directories

Path Synopsis
Package mock_wsclient is a generated GoMock package.
 Package mock_wsclient is a generated GoMock package.
utils
mock
Package mock_wsconn is a generated GoMock package.
 Package mock_wsconn is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.