clients

Documentation

Overview

package clients is a wrapper around grpc clients so clients can work with non protobuf structs and handle unmarshaling

Index

• func DownloadPluginFromGithub(ctx context.Context, localPath string, org string, name string, version string, ...) error
• func WithDestinationDirectory(directory string) func(*DestinationClient)
• func WithDestinationGrpcConn(userConn *grpc.ClientConn) func(*DestinationClient)
• func WithDestinationLogger(logger zerolog.Logger) func(*DestinationClient)
• func WithSourceDirectory(directory string) func(*SourceClient)
• func WithSourceGRPCConnection(userConn *grpc.ClientConn) func(*SourceClient)
• func WithSourceLogger(logger zerolog.Logger) func(*SourceClient)
• type DestinationClient
  • func NewDestinationClient(ctx context.Context, registry specs.Registry, path string, version string, ...) (*DestinationClient, error)
  • func (c *DestinationClient) Close(ctx context.Context) error
  • func (c *DestinationClient) DeleteStale(ctx context.Context, tables schema.Tables, source string, timestamp time.Time) error
  • func (c *DestinationClient) GetProtocolVersion(ctx context.Context) (uint64, error)
  • func (c *DestinationClient) GetWaitError() error
  • func (c *DestinationClient) Initialize(ctx context.Context, spec specs.Destination) error
  • func (c *DestinationClient) Migrate(ctx context.Context, tables []*schema.Table) error
  • func (c *DestinationClient) Name(ctx context.Context) (string, error)
  • func (c *DestinationClient) Terminate() error
  • func (c *DestinationClient) Version(ctx context.Context) (string, error)
  • func (c *DestinationClient) Write(ctx context.Context, source string, syncTime time.Time, ...) (uint64, error)
• type DestinationClientOption
• type FetchResultMessage
• type PluginType
• type SourceClient
  • func NewSourceClient(ctx context.Context, registry specs.Registry, path string, version string, ...) (*SourceClient, error)
  • func (c *SourceClient) GetProtocolVersion(ctx context.Context) (uint64, error)
  • func (c *SourceClient) GetSyncSummary(ctx context.Context) (*schema.SyncSummary, error)
  • func (c *SourceClient) GetTables(ctx context.Context) ([]*schema.Table, error)
  • func (c *SourceClient) GetWaitError() error
  • func (c *SourceClient) Name(ctx context.Context) (string, error)
  • func (c *SourceClient) Sync(ctx context.Context, spec specs.Source, res chan<- []byte) error
  • func (c *SourceClient) Terminate() error
  • func (c *SourceClient) Version(ctx context.Context) (string, error)
• type SourceClientOption

Functions

func DownloadPluginFromGithub

func DownloadPluginFromGithub(ctx context.Context, localPath string, org string, name string, version string, typ PluginType) error

func WithDestinationDirectory

func WithDestinationDirectory(directory string) func(*DestinationClient)

func WithDestinationGrpcConn

func WithDestinationGrpcConn(userConn *grpc.ClientConn) func(*DestinationClient)

func WithDestinationLogger

func WithDestinationLogger(logger zerolog.Logger) func(*DestinationClient)

func WithSourceDirectory

func WithSourceDirectory(directory string) func(*SourceClient)

func WithSourceGRPCConnection

func WithSourceGRPCConnection(userConn *grpc.ClientConn) func(*SourceClient)

func WithSourceLogger

func WithSourceLogger(logger zerolog.Logger) func(*SourceClient)

Types

type DestinationClient

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

func NewDestinationClient

func NewDestinationClient(ctx context.Context, registry specs.Registry, path string, version string, opts ...DestinationClientOption) (*DestinationClient, error)

func (*DestinationClient) Close

func (c *DestinationClient) Close(ctx context.Context) error

func (*DestinationClient) DeleteStale

func (c *DestinationClient) DeleteStale(ctx context.Context, tables schema.Tables, source string, timestamp time.Time) error

func (*DestinationClient) GetProtocolVersion

func (c *DestinationClient) GetProtocolVersion(ctx context.Context) (uint64, error)

func (*DestinationClient) GetWaitError

func (c *DestinationClient) GetWaitError() error

func (*DestinationClient) Initialize

func (c *DestinationClient) Initialize(ctx context.Context, spec specs.Destination) error

func (*DestinationClient) Migrate

func (c *DestinationClient) Migrate(ctx context.Context, tables []*schema.Table) error

func (*DestinationClient) Name

func (c *DestinationClient) Name(ctx context.Context) (string, error)

func (*DestinationClient) Terminate

func (c *DestinationClient) Terminate() error

Terminate is used only in conjunction with NewManagedDestinationClient. It closes the connection it created, kills the spawned process and removes the socket file.

func (*DestinationClient) Version

func (c *DestinationClient) Version(ctx context.Context) (string, error)

func (*DestinationClient) Write

func (c *DestinationClient) Write(ctx context.Context, source string, syncTime time.Time, resources <-chan []byte) (uint64, error)

Write writes rows as they are received from the channel to the destination plugin. resources is marshaled schema.Resource. We are not marshalling this inside the function because usually it is alreadun marshalled from the destination plugin.

type DestinationClientOption

type DestinationClientOption func(*DestinationClient)

type FetchResultMessage

type FetchResultMessage struct {
	Resource []byte
}

type PluginType

type PluginType string

const (
	PluginTypeSource      PluginType = "source"
	PluginTypeDestination PluginType = "destination"
	DefaultDownloadDir               = ".cq"
)

type SourceClient

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

SourceClient

func NewSourceClient

func NewSourceClient(ctx context.Context, registry specs.Registry, path string, version string, opts ...SourceClientOption) (*SourceClient, error)

NewSourceClient connect to gRPC server running source plugin and returns a new SourceClient

func (*SourceClient) GetProtocolVersion

func (c *SourceClient) GetProtocolVersion(ctx context.Context) (uint64, error)

func (*SourceClient) GetSyncSummary

func (c *SourceClient) GetSyncSummary(ctx context.Context) (*schema.SyncSummary, error)

func (*SourceClient) GetTables

func (c *SourceClient) GetTables(ctx context.Context) ([]*schema.Table, error)

func (*SourceClient) GetWaitError

func (c *SourceClient) GetWaitError() error

func (*SourceClient) Name

func (c *SourceClient) Name(ctx context.Context) (string, error)

func (*SourceClient) Sync

func (c *SourceClient) Sync(ctx context.Context, spec specs.Source, res chan<- []byte) error

Sync start syncing for the source client per the given spec and returning the results in the given channel. res is marshaled schema.Resource. We are not unmarshalling this for performance reasons as usually this is sent over-the-wire anyway to a source plugin

func (*SourceClient) Terminate

func (c *SourceClient) Terminate() error

Terminate is used only in conjunction with NewManagedSourceClient. It closes the connection it created, kills the spawned process and removes the socket file.

func (*SourceClient) Version

func (c *SourceClient) Version(ctx context.Context) (string, error)

type SourceClientOption

type SourceClientOption func(*SourceClient)