adapter

Documentation

Overview

Package adapter provides a framework for implementing protocol adapters for Spanner.

Index

• Variables
• func ExtractRetryDelay(err error) (time.Duration, bool)
• func MockAdaptMessageGrpc(returnResponsesInChunks bool)
• func MockCreateSessionGrpc(mock_session_names ...string)
• func NewDefaultGlobalState(size int) (*globalState, error)
• func ResetGrpcFuncs() func()
• func RunAdaptMessageWithRetry(ctx context.Context, disableRetry bool, ...) (adapterpb.Adapter_AdaptMessageClient, error)
• func RunCreateAdapterSessionWithRetry(ctx context.Context, f func(context.Context) error) error
• type AdapterClient
• type GrpcFuncs
• type Mock_Cassandra_AdaptMessageClient
  • func (mc *Mock_Cassandra_AdaptMessageClient) CloseSend() error
  • func (mc *Mock_Cassandra_AdaptMessageClient) Context() context.Context
  • func (mc *Mock_Cassandra_AdaptMessageClient) Header() (metadata.MD, error)
  • func (mc *Mock_Cassandra_AdaptMessageClient) Recv() (*adapterpb.AdaptMessageResponse, error)
  • func (mc *Mock_Cassandra_AdaptMessageClient) RecvMsg(m any) error
  • func (mc *Mock_Cassandra_AdaptMessageClient) SendMsg(m any) error
  • func (mc *Mock_Cassandra_AdaptMessageClient) Trailer() metadata.MD
• type Options
• type Protocol
• type TCPProxy
  • func NewTCPProxy(opts Options) (*TCPProxy, error)
  • func (proxy *TCPProxy) Addr() net.Addr
  • func (proxy *TCPProxy) Close()

Variables

var (
	// SessionRefreshTimeInterval defines the interval for refreshing Adapter
	// sessions. Adapter Sessions have a 7-day lifetime and are refreshed 1 day
	// before expiry to provide a buffer against potential delays.
	SessionRefreshTimeInterval = 6 * 24 * time.Hour
	CreateSessionGrpc          = func(ctx context.Context, req *adapterpb.CreateSessionRequest, cl *AdapterClient) (*adapterpb.Session, error) {
		var md metadata.MD
		resp, err := cl.gapicClient.CreateSession(
			ctx,
			req,
			gax.WithGRPCOptions(grpc.Header(&md)),
		)
		if err != nil {
			return nil, err
		}
		return resp, nil
	}
)

var (
	AdaptMessageGrpc = func(
		ctx context.Context,
		req *adapterpb.AdaptMessageRequest,
		cl *AdapterClient,
	) (adapterpb.Adapter_AdaptMessageClient, error) {
		var md metadata.MD
		request, err := cl.gapicClient.AdaptMessage(
			ctx,
			req,
			gax.WithGRPCOptions(grpc.Header(&md)),
		)
		if err != nil {
			return nil, err
		}
		return request, nil
	}
)

var DefaultRetryBackoff = gax.Backoff{
	Initial:    20 * time.Millisecond,
	Max:        32 * time.Second,
	Multiplier: 1.3,
}

DefaultRetryBackoff is used for retryers as a fallback value when the server did not return any retry information.

var (

	// api options to disable application default credential in unit test
	SkipAuthOpts = []option.ClientOption{
		option.WithEndpoint("localhost:443"),
		option.WithGRPCDialOption(
			grpc.WithTransportCredentials(insecure.NewCredentials()),
		),
		option.WithoutAuthentication(),
		internaloption.SkipDialSettingsValidation(),
	}
)

Functions

func ExtractRetryDelay

func ExtractRetryDelay(err error) (time.Duration, bool)

ExtractRetryDelay extracts retry backoff from a grpc error if present.

func MockAdaptMessageGrpc

func MockAdaptMessageGrpc(returnResponsesInChunks bool)

func MockCreateSessionGrpc

func MockCreateSessionGrpc(mock_session_names ...string)

func NewDefaultGlobalState

func NewDefaultGlobalState(size int) (*globalState, error)

NewDefaultGlobalState creates a new default prepared cache capping the max item capacity to `size`.

func ResetGrpcFuncs

func ResetGrpcFuncs() func()

func RunAdaptMessageWithRetry

func RunAdaptMessageWithRetry(
	ctx context.Context,
	disableRetry bool,
	f func(ctx context.Context) (adapterpb.Adapter_AdaptMessageClient, error),
) (adapterpb.Adapter_AdaptMessageClient, error)

RunAdaptMessageWithRetry executes the provided function with a retry mechanism based on the given policy.

func RunCreateAdapterSessionWithRetry

func RunCreateAdapterSessionWithRetry(
	ctx context.Context,
	f func(context.Context) error,
) error

RunFuncWithRetry executes the provided function with a retry mechanism based on the given policy.

Types

type AdapterClient

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

The adapterClient encapsulates the gRPC connection / adapter stub creation. It is also responsible for refreshing the multiplexed session.

type GrpcFuncs

type GrpcFuncs struct {
	CreateSession func(ctx context.Context, req *adapterpb.CreateSessionRequest, cl *AdapterClient) (*adapterpb.Session, error)
	AdaptMessage  func(
		ctx context.Context,
		req *adapterpb.AdaptMessageRequest,
		cl *AdapterClient,
	) (adapterpb.Adapter_AdaptMessageClient, error)
}

type Mock_Cassandra_AdaptMessageClient

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

func (*Mock_Cassandra_AdaptMessageClient) CloseSend

func (mc *Mock_Cassandra_AdaptMessageClient) CloseSend() error

func (*Mock_Cassandra_AdaptMessageClient) Context

func (mc *Mock_Cassandra_AdaptMessageClient) Context() context.Context

func (*Mock_Cassandra_AdaptMessageClient) Header

func (mc *Mock_Cassandra_AdaptMessageClient) Header() (metadata.MD, error)

func (*Mock_Cassandra_AdaptMessageClient) Recv

func (mc *Mock_Cassandra_AdaptMessageClient) Recv() (*adapterpb.AdaptMessageResponse, error)

func (*Mock_Cassandra_AdaptMessageClient) RecvMsg

func (mc *Mock_Cassandra_AdaptMessageClient) RecvMsg(m any) error

func (*Mock_Cassandra_AdaptMessageClient) SendMsg

func (mc *Mock_Cassandra_AdaptMessageClient) SendMsg(m any) error

func (*Mock_Cassandra_AdaptMessageClient) Trailer

func (mc *Mock_Cassandra_AdaptMessageClient) Trailer() metadata.MD

type Options

type Options struct {
	// Spanner database uri to connect to.
	DatabaseUri string
	// Optional Spanner service endpoint. Defaults to spanner.googleapis.com:443
	SpannerEndpoint string
	// Protocol type (ie: cassandra).
	Protocol Protocol
	// Number of channels when dial grpc connection. Defaults to 4.
	NumGrpcChannels int
	// Optional Endpoint to start TCP server. Defaults to localhost:9042
	TCPEndpoint string
	// Optional boolean indicate whether to disable automatic grpc retry for
	// AdaptMessage API. Defauls to false.
	DisableAdaptMessageRetry bool
	// The maximum delay in milliseconds. Default is 0 (disabled).
	MaxCommitDelay int
	// Optional google api opts. Default to empty.
	GoogleApiOpts []option.ClientOption
}

Options for configuring the adapter.

type Protocol

type Protocol interface {

	// Returns the protocol identifier.
	Name() string
}

Protocol is the interface that all protocols must implement.

type TCPProxy

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

TCPProxy encapsulates a Spanner Adapter proxy.

func NewTCPProxy

func NewTCPProxy(opts Options) (*TCPProxy, error)

NewTCPProxy returns a new Spanner Adapter proxy.

func (*TCPProxy) Addr

func (proxy *TCPProxy) Addr() net.Addr

Addr returns the address of the proxy.

func (*TCPProxy) Close

func (proxy *TCPProxy) Close()

Close closes the proxy.