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, githubPath string, version string, ...) error
• func WithDestinationLogger(logger zerolog.Logger) func(*DestinationClient)
• func WithSourceLogger(logger zerolog.Logger) func(*SourceClient)
• type DestinationClient
  • func NewDestinationClient(cc grpc.ClientConnInterface) *DestinationClient
  • func NewManagedDestinationClient(ctx context.Context, path string, opts ...DestinationClientOption) (*DestinationClient, error)
  • func (c *DestinationClient) Close() 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) Version(ctx context.Context) (string, error)
  • func (c *DestinationClient) Write(ctx context.Context, resources <-chan []byte) (uint64, error)
• type DestinationClientOption
• type FetchResultMessage
• type PluginType
• type SourceClient
  • func NewManagedSourceClient(ctx context.Context, path string, opts ...SourceClientOption) (*SourceClient, error)
  • func NewSourceClient(cc grpc.ClientConnInterface) *SourceClient
  • func (c *SourceClient) Close() 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) Version(ctx context.Context) (string, error)
• type SourceClientOption

Functions

func DownloadPluginFromGithub

func DownloadPluginFromGithub(ctx context.Context, localPath string, githubPath string, version string, typ PluginType, writers ...io.Writer) error

func WithDestinationLogger

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

func WithSourceLogger

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

Types

type DestinationClient

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

func NewDestinationClient

func NewDestinationClient(cc grpc.ClientConnInterface) *DestinationClient

func NewManagedDestinationClient

func NewManagedDestinationClient(ctx context.Context, path string, opts ...DestinationClientOption) (*DestinationClient, error)

NewManagedDestinationClient starts a new destination plugin process, connects to it via gRPC server and returns a new DestinationClient

func (*DestinationClient) Close

func (c *DestinationClient) Close() error

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

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) Version

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

func (*DestinationClient) Write

func (c *DestinationClient) Write(ctx context.Context, 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 source 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"
)

type SourceClient

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

SourceClient

func NewManagedSourceClient

func NewManagedSourceClient(ctx context.Context, path string, opts ...SourceClientOption) (*SourceClient, error)

NewManagedSourceClient starts a new source plugin process, connects to it via gRPC server and returns a new SourceClient

func NewSourceClient

func NewSourceClient(cc grpc.ClientConnInterface) *SourceClient

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

func (*SourceClient) Close

func (c *SourceClient) Close() error

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

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 destination plugin

func (*SourceClient) Version

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

type SourceClientOption

type SourceClientOption func(*SourceClient)