sdk

Oct 28, 2026

Changes in this version

+ const CodeBatchDrainSemanticsPending

+ const CodeBatchSizeDefaulted

+ const CodeBatchUnboundedWait

+ func CodeFromMessage(msg string) (string, bool)

Oct 15, 2026

Changes in this version

+ const AcceptanceOutcomeFailed

+ const AcceptanceOutcomePassed

+ const AcceptanceOutcomeSkipped

+ const AcceptanceReportEnv

+ const AcceptanceSuiteV2Revision

+ const CertificationMinBatchSize

+ const CertificationMinRecordBytes

+ const DefaultMaxRedelivery

+ const MaxAcceptedRedelivery

+ const ReportSchemaVersion

+ const SkipDeclared

+ const SkipDriver

+ const SkipInCase

+ const SkipNotApplicable

+ const SkipReplaced

+ const SkipUnclaimed

+ func AcceptanceTestV2(t *testing.T, driver AcceptanceTestDriver)

+ type AcceptanceCapabilities struct

+ Destination DestinationCapabilities

+ MaxBatchSize int

+ MaxRecordBytes int

+ Source SourceCapabilities

+ ThroughputDuration time.Duration

+ Waivers []string

+ type AcceptanceCapabilitiesProvider interface

+ AcceptanceCapabilities func() AcceptanceCapabilities

+ type AcceptanceCaseResult struct

+ DurationSeconds float64

+ ID string

+ Name string

+ Outcome string

+ Reason string

+ SkipKind string

+ type AcceptanceKeyedRecordGenerator interface

+ GenerateRecordWithKey func(t *testing.T, op opencdc.Operation, key string) opencdc.Record

+ type AcceptanceLargeRecordGenerator interface

+ GenerateLargeRecord func(t *testing.T, op opencdc.Operation, size int) opencdc.Record

+ type AcceptanceMetric struct

+ Detail string

+ Name string

+ Unit string

+ Value float64

+ type AcceptanceProcessDeathDriver interface

+ KilledSessionID func(*testing.T) string

+ ReleaseSession func(ctx context.Context, sessionID string) error

+ type AcceptanceReadAllDriver interface

+ ReadAllFromDestination func(*testing.T) []opencdc.Record

+ type AcceptanceReport struct

+ Build BuildSettings

+ Capabilities AcceptanceCapabilities

+ CaseContentHash string

+ CaseSetHash string

+ Cases []AcceptanceCaseResult

+ Complete bool

+ Connector struct{ ... }

+ DurationSeconds float64

+ Exceptions []string

+ ExpectedCaseCount int

+ ExpectedCaseIDs []string

+ Finished bool

+ Goleak struct{ ... }

+ Limits struct{ ... }

+ Metrics []AcceptanceMetric

+ Passed bool

+ ProcessDeathHookCalls int

+ ReportSchemaVersion int

+ RunID string

+ SDKVersion string

+ SDKVersionSource string

+ StartedAt time.Time

+ SuiteRevision string

+ SuiteSourceHash string

+ SuiteVersion int

+ TestBinarySHA256 string

+ UnexplainedSkips int

+ type BuildSettings struct

+ CGO bool

+ Compiler string

+ GOARCH string

+ GOOS string

+ GoVersion string

+ Race bool

+ Tags string

+ Trimpath bool

type ConfigurableAcceptanceTestDriver

+ func (d ConfigurableAcceptanceTestDriver) AcceptanceCapabilities() AcceptanceCapabilities

+ func (d ConfigurableAcceptanceTestDriver) GoleakIgnoredFunctions() []string

+ func (d ConfigurableAcceptanceTestDriver) GoleakOtherOptionCount() int

+ func (d ConfigurableAcceptanceTestDriver) ReadAllFromSource(t *testing.T) ([]opencdc.Record, bool)

type ConfigurableAcceptanceTestDriverConfig

+ Capabilities AcceptanceCapabilities

+ GoleakIgnoreFunctions []string

+ ReadAllFromSource bool

+ type DestinationCapabilities struct

+ Idempotence IdempotencyClaim

+ OrdersByKey bool

+ SchemaEvolution SchemaEvolutionPolicy

+ WriteOnly bool

+ type IdempotencyClaim string

+ const IdempotenceNo

+ const IdempotenceUnspecified

+ const IdempotenceYes

+ type SchemaEvolutionPolicy string

+ const SchemaEvolutionEvolve

+ const SchemaEvolutionNone

+ const SchemaEvolutionReject

+ type SourceCapabilities struct

+ ExactResume bool

+ MaxRedelivery int

+ OrdersByKey bool

+ SchemaEvolution SchemaEvolutionPolicy

+ Synthetic bool

Jan 31, 2025

Changes in this version

+ func YAMLSpecification(rawYaml, version string) func() Specification

type DefaultDestinationMiddleware

+ func (c *DefaultDestinationMiddleware) Validate(ctx context.Context) error

type DefaultSourceMiddleware

+ func (c *DefaultSourceMiddleware) Validate(ctx context.Context) error

type Destination

+ Config func() DestinationConfig

+ type DestinationConfig interface

type DestinationWithBatch

+ BatchDelay time.Duration

+ BatchSize int

type DestinationWithRateLimit

+ Burst int

+ RatePerSecond float64

type DestinationWithRecordFormat

+ RecordFormat *string

+ RecordFormatOptions string

+ RecordSerializers []RecordSerializer

+ func (c *DestinationWithRecordFormat) DefaultRecordSerializers() []RecordSerializer

+ func (c *DestinationWithRecordFormat) Validate(context.Context) error

type DestinationWithSchemaExtraction

+ KeyEnabled *bool

+ PayloadEnabled *bool

type Source

+ Config func() SourceConfig

+ type SourceConfig interface

type SourceWithBatch

+ BatchDelay *time.Duration

+ BatchSize *int

type SourceWithSchemaContext

+ Enabled *bool

+ Name *string

type SourceWithSchemaExtraction

+ KeyEnabled *bool

+ KeySubject *string

+ PayloadEnabled *bool

+ PayloadSubject *string

+ SchemaTypeStr string

+ func (c *SourceWithSchemaExtraction) SchemaType() schema.Type

type Specification

+ func ParseYAMLSpecification(ctx context.Context, rawYaml, version string) (Specification, error)

type UnimplementedDestination

+ func (UnimplementedDestination) Config() DestinationConfig

+ type UnimplementedDestinationConfig struct

+ func (UnimplementedDestinationConfig) Validate(context.Context) error

type UnimplementedSource

+ func (UnimplementedSource) Config() SourceConfig

+ type UnimplementedSourceConfig struct

+ func (UnimplementedSourceConfig) Validate(context.Context) error

+ type Validatable interface

+ Validate func(context.Context) error

Nov 5, 2024

Changes in this version

type Source

+ ReadN func(context.Context, int) ([]opencdc.Record, error)

+ type SourceWithBatch struct

+ Config SourceWithBatchConfig

+ func (s *SourceWithBatch) Wrap(impl Source) Source

+ type SourceWithBatchConfig struct

+ BatchDelay *time.Duration

+ BatchSize *int

+ func (c SourceWithBatchConfig) Apply(m SourceMiddleware)

+ func (c SourceWithBatchConfig) BatchDelayParameterName() string

+ func (c SourceWithBatchConfig) BatchSizeParameterName() string

+ type SourceWithEncoding struct

+ func (s SourceWithEncoding) Apply(SourceMiddleware)

+ func (s SourceWithEncoding) Wrap(impl Source) Source

type UnimplementedSource

+ func (s UnimplementedSource) ReadN(context.Context, int) ([]opencdc.Record, error)

Sep 16, 2024

Changes in this version

+ func ConnectorIDFromContext(ctx context.Context) string

Aug 7, 2024

Changes in this version

+ type DestinationMiddlewareOption interface

+ Apply func(DestinationMiddleware)

type DestinationWithBatch

+ Config DestinationWithBatchConfig

+ type DestinationWithBatchConfig struct

+ BatchDelay time.Duration

+ BatchSize int

+ func (c DestinationWithBatchConfig) Apply(m DestinationMiddleware)

+ func (c DestinationWithBatchConfig) BatchDelayParameterName() string

+ func (c DestinationWithBatchConfig) BatchSizeParameterName() string

type DestinationWithRateLimit

+ Config DestinationWithRateLimitConfig

+ type DestinationWithRateLimitConfig struct

+ Burst int

+ RatePerSecond float64

+ func (c DestinationWithRateLimitConfig) Apply(m DestinationMiddleware)

+ func (c DestinationWithRateLimitConfig) RateBurstParameterName() string

+ func (c DestinationWithRateLimitConfig) RatePerSecondParameterName() string

type DestinationWithRecordFormat

+ Config DestinationWithRecordFormatConfig

+ type DestinationWithRecordFormatConfig struct

+ DefaultRecordFormat string

+ RecordSerializers []RecordSerializer

+ func (c DestinationWithRecordFormatConfig) Apply(m DestinationMiddleware)

+ func (c DestinationWithRecordFormatConfig) DefaultRecordSerializers() []RecordSerializer

+ func (c DestinationWithRecordFormatConfig) RecordFormatOptionsParameterName() string

+ func (c DestinationWithRecordFormatConfig) RecordFormatParameterName() string

+ type DestinationWithSchemaExtraction struct

+ Config DestinationWithSchemaExtractionConfig

+ func (s *DestinationWithSchemaExtraction) Wrap(impl Destination) Destination

+ type DestinationWithSchemaExtractionConfig struct

+ KeyEnabled *bool

+ PayloadEnabled *bool

+ func (c DestinationWithSchemaExtractionConfig) Apply(m DestinationMiddleware)

+ func (c DestinationWithSchemaExtractionConfig) SchemaKeyEnabledParameterName() string

+ func (c DestinationWithSchemaExtractionConfig) SchemaPayloadEnabledParameterName() string

+ type GenericRecordSerializer struct

+ func (rf GenericRecordSerializer) Configure(optRaw string) (RecordSerializer, error)

+ func (rf GenericRecordSerializer) Name() string

+ func (rf GenericRecordSerializer) Serialize(r opencdc.Record) ([]byte, error)

+ type RecordSerializer interface

+ Configure func(string) (RecordSerializer, error)

+ Name func() string

+ type SourceMiddlewareOption interface

+ Apply func(SourceMiddleware)

+ type SourceWithSchemaContext struct

+ Config SourceWithSchemaContextConfig

+ func (s *SourceWithSchemaContext) Wrap(impl Source) Source

+ type SourceWithSchemaContextConfig struct

+ Enabled *bool

+ Name *string

+ func (c SourceWithSchemaContextConfig) Apply(m SourceMiddleware)

+ func (c SourceWithSchemaContextConfig) EnabledParameterName() string

+ func (c SourceWithSchemaContextConfig) NameParameterName() string

+ type SourceWithSchemaExtraction struct

+ Config SourceWithSchemaExtractionConfig

+ func (s *SourceWithSchemaExtraction) Wrap(impl Source) Source

+ type SourceWithSchemaExtractionConfig struct

+ KeyEnabled *bool

+ KeySubject *string

+ PayloadEnabled *bool

+ PayloadSubject *string

+ SchemaType schema.Type

+ func (c SourceWithSchemaExtractionConfig) Apply(m SourceMiddleware)

+ func (c SourceWithSchemaExtractionConfig) SchemaKeyEnabledParameterName() string

+ func (c SourceWithSchemaExtractionConfig) SchemaKeySubjectParameterName() string

+ func (c SourceWithSchemaExtractionConfig) SchemaPayloadEnabledParameterName() string

+ func (c SourceWithSchemaExtractionConfig) SchemaPayloadSubjectParameterName() string

+ func (c SourceWithSchemaExtractionConfig) SchemaTypeParameterName() string

+ type TemplateRecordSerializer struct

+ func (e TemplateRecordSerializer) Configure(tmpl string) (RecordSerializer, error)

+ func (e TemplateRecordSerializer) Name() string

+ func (e TemplateRecordSerializer) Serialize(r opencdc.Record) ([]byte, error)

Apr 18, 2024

Changes in this version

+ const MetadataCollection

+ const MetadataConduitDLQNackError

+ const MetadataConduitDLQNackNodeID

+ const MetadataConduitSourceConnectorID

+ var ErrInvalidParameterType = config.ErrInvalidParameterType

+ var ErrInvalidParameterValue = config.ErrInvalidParameterValue

+ var ErrInvalidValidationType = config.ErrInvalidValidationType

+ var ErrUnrecognizedParameter = config.ErrUnrecognizedParameter

+ func BenchmarkSource(b *testing.B, s Source, cfg map[string]string)

type AcceptanceTestDriver

+ Context func() context.Context

type ConfigurableAcceptanceTestDriver

+ func (d ConfigurableAcceptanceTestDriver) Context() context.Context

type ConfigurableAcceptanceTestDriverConfig

+ Context context.Context

type Metadata

+ func (m Metadata) GetCollection() (string, error)

+ func (m Metadata) GetConduitDLQNackError() (string, error)

+ func (m Metadata) GetConduitDLQNackNodeID() (string, error)

+ func (m Metadata) GetConduitSourceConnectorID() (string, error)

+ func (m Metadata) SetCollection(collection string)

+ func (m Metadata) SetConduitDLQNackError(err string)

+ func (m Metadata) SetConduitDLQNackNodeID(id string)

+ func (m Metadata) SetConduitSourceConnectorID(id string)

Apr 11, 2023

Changes in this version

type Destination

+ LifecycleOnCreated func(ctx context.Context, config map[string]string) error

+ LifecycleOnDeleted func(ctx context.Context, config map[string]string) error

+ LifecycleOnUpdated func(ctx context.Context, configBefore, configAfter map[string]string) error

type Source

+ LifecycleOnCreated func(ctx context.Context, config map[string]string) error

+ LifecycleOnDeleted func(ctx context.Context, config map[string]string) error

+ LifecycleOnUpdated func(ctx context.Context, configBefore, configAfter map[string]string) error

type UnimplementedDestination

+ func (UnimplementedDestination) LifecycleOnCreated(context.Context, map[string]string) error

+ func (UnimplementedDestination) LifecycleOnDeleted(context.Context, map[string]string) error

+ func (UnimplementedDestination) LifecycleOnUpdated(context.Context, map[string]string, map[string]string) error

type UnimplementedSource

+ func (UnimplementedSource) LifecycleOnCreated(context.Context, map[string]string) error

+ func (UnimplementedSource) LifecycleOnDeleted(context.Context, map[string]string) error

+ func (UnimplementedSource) LifecycleOnUpdated(context.Context, map[string]string, map[string]string) error

Feb 14, 2023

Changes in this version

type DebeziumConverter

+ RawDataKey string

Jan 31, 2023

Changes in this version

+ type Converter interface

+ Configure func(map[string]string) (Converter, error)

+ Convert func(Record) (any, error)

+ Name func() string

+ type DebeziumConverter struct

+ SchemaName string

+ func (c DebeziumConverter) Configure(opt map[string]string) (Converter, error)

+ func (c DebeziumConverter) Convert(r Record) (any, error)

+ func (c DebeziumConverter) Name() string

type DestinationWithBatch

+ func (d DestinationWithBatch) BatchDelayParameterName() string

+ func (d DestinationWithBatch) BatchSizeParameterName() string

type DestinationWithRateLimit

+ func (d DestinationWithRateLimit) RateBurstParameterName() string

+ func (d DestinationWithRateLimit) RatePerSecondParameterName() string

+ type DestinationWithRecordFormat struct

+ DefaultRecordFormat string

+ RecordFormatters []RecordFormatter

+ func (d DestinationWithRecordFormat) DefaultRecordFormatters() []RecordFormatter

+ func (d DestinationWithRecordFormat) RecordFormatOptionsParameterName() string

+ func (d DestinationWithRecordFormat) RecordFormatParameterName() string

+ func (d DestinationWithRecordFormat) Wrap(impl Destination) Destination

+ type Encoder interface

+ Configure func(options map[string]string) (Encoder, error)

+ Encode func(r any) ([]byte, error)

+ Name func() string

+ type GenericRecordFormatter struct

+ func (rf GenericRecordFormatter) Configure(optRaw string) (RecordFormatter, error)

+ func (rf GenericRecordFormatter) Format(r Record) ([]byte, error)

+ func (rf GenericRecordFormatter) Name() string

+ type JSONEncoder struct

+ func (e JSONEncoder) Configure(map[string]string) (Encoder, error)

+ func (e JSONEncoder) Encode(v any) ([]byte, error)

+ func (e JSONEncoder) Name() string

+ type OpenCDCConverter struct

+ func (c OpenCDCConverter) Configure(map[string]string) (Converter, error)

+ func (c OpenCDCConverter) Convert(r Record) (any, error)

+ func (c OpenCDCConverter) Name() string

+ type RecordFormatter interface

+ Configure func(string) (RecordFormatter, error)

+ Format func(Record) ([]byte, error)

+ Name func() string

+ type TemplateRecordFormatter struct

+ func (e TemplateRecordFormatter) Configure(tmpl string) (RecordFormatter, error)

+ func (e TemplateRecordFormatter) Format(r Record) ([]byte, error)

+ func (e TemplateRecordFormatter) Name() string

Dec 15, 2022

Changes in this version

+ const MetadataConduitDestinationPluginName

+ const MetadataConduitDestinationPluginVersion

+ const MetadataConduitSourcePluginName

+ const MetadataConduitSourcePluginVersion

+ var ErrExclusionValidationFail = errors.New("exclusion validation failed")

+ var ErrGreaterThanValidationFail = errors.New("greater-than validation failed")

+ var ErrInclusionValidationFail = errors.New("inclusion validation failed")

+ var ErrInvalidParamType = errors.New("invalid parameter type")

+ var ErrInvalidParamValue = errors.New("invalid parameter value")

+ var ErrLessThanValidationFail = errors.New("less-than validation failed")

+ var ErrRegexValidationFail = errors.New("regex validation failed")

+ var ErrRequiredParameterMissing = errors.New("required parameter is not provided")

type Metadata

+ func (m Metadata) GetConduitDestinationPluginName() (string, error)

+ func (m Metadata) GetConduitDestinationPluginVersion() (string, error)

+ func (m Metadata) GetConduitSourcePluginName() (string, error)

+ func (m Metadata) GetConduitSourcePluginVersion() (string, error)

+ func (m Metadata) SetConduitDestinationPluginName(name string)

+ func (m Metadata) SetConduitDestinationPluginVersion(version string)

+ func (m Metadata) SetConduitSourcePluginName(name string)

+ func (m Metadata) SetConduitSourcePluginVersion(version string)

type Parameter

+ Type ParameterType

+ Validations []Validation

+ type ParameterType int

+ const ParameterTypeBool

+ const ParameterTypeDuration

+ const ParameterTypeFile

+ const ParameterTypeFloat

+ const ParameterTypeInt

+ const ParameterTypeString

+ type Validation interface

+ type ValidationExclusion struct

+ List []string

+ type ValidationGreaterThan struct

+ Value float64

+ type ValidationInclusion struct

+ List []string

+ type ValidationLessThan struct

+ Value float64

+ type ValidationRegex struct

+ Regex *regexp.Regexp

+ type ValidationRequired struct

Sep 27, 2022

Changes in this version

+ const MetadataConduitPluginName

+ const MetadataConduitPluginVersion

+ const MetadataCreatedAt

+ const MetadataOpenCDCVersion

+ const MetadataReadAt

+ var ErrMetadataFieldNotFound = errors.New("metadata field not found")

+ var Util struct

+ func AcceptanceTest(t *testing.T, driver AcceptanceTestDriver)

+ type AcceptanceTestDriver interface

+ AfterTest func(*testing.T)

+ BeforeTest func(*testing.T)

+ Connector func() Connector

+ DestinationConfig func(*testing.T) map[string]string

+ GenerateRecord func(*testing.T, Operation) Record

+ GoleakOptions func(*testing.T) []goleak.Option

+ ReadFromDestination func(*testing.T, []Record) []Record

+ ReadTimeout func() time.Duration

+ SourceConfig func(*testing.T) map[string]string

+ WriteTimeout func() time.Duration

+ WriteToSource func(*testing.T, []Record) []Record

+ type Change struct

+ After Data

+ Before Data

+ type ConfigurableAcceptanceTestDriver struct

+ Config ConfigurableAcceptanceTestDriverConfig

+ func (d ConfigurableAcceptanceTestDriver) AfterTest(t *testing.T)

+ func (d ConfigurableAcceptanceTestDriver) BeforeTest(t *testing.T)

+ func (d ConfigurableAcceptanceTestDriver) Connector() Connector

+ func (d ConfigurableAcceptanceTestDriver) DestinationConfig(*testing.T) map[string]string

+ func (d ConfigurableAcceptanceTestDriver) GenerateData(t *testing.T) Data

+ func (d ConfigurableAcceptanceTestDriver) GenerateRecord(t *testing.T, op Operation) Record

+ func (d ConfigurableAcceptanceTestDriver) GenerateValue(t *testing.T) interface{}

+ func (d ConfigurableAcceptanceTestDriver) GoleakOptions(_ *testing.T) []goleak.Option

+ func (d ConfigurableAcceptanceTestDriver) ReadFromDestination(t *testing.T, records []Record) []Record

+ func (d ConfigurableAcceptanceTestDriver) ReadTimeout() time.Duration

+ func (d ConfigurableAcceptanceTestDriver) Skip(t *testing.T)

+ func (d ConfigurableAcceptanceTestDriver) SourceConfig(*testing.T) map[string]string

+ func (d ConfigurableAcceptanceTestDriver) WriteTimeout() time.Duration

+ func (d ConfigurableAcceptanceTestDriver) WriteToSource(t *testing.T, records []Record) []Record

+ type ConfigurableAcceptanceTestDriverConfig struct

+ AfterTest func(t *testing.T)

+ BeforeTest func(t *testing.T)

+ Connector Connector

+ DestinationConfig map[string]string

+ GenerateDataType GenerateDataType

+ GoleakOptions []goleak.Option

+ ReadTimeout time.Duration

+ Skip []string

+ SourceConfig map[string]string

+ WriteTimeout time.Duration

+ type Connector struct

+ NewDestination func() Destination

+ NewSource func() Source

+ NewSpecification func() Specification

type Destination

+ func DestinationWithMiddleware(d Destination, middleware ...DestinationMiddleware) Destination

+ Parameters func() map[string]Parameter

+ type DestinationMiddleware interface

+ Wrap func(Destination) Destination

+ func DefaultDestinationMiddleware() []DestinationMiddleware

+ type DestinationUtil struct

+ func (DestinationUtil) Route(ctx context.Context, rec Record, ...) error

+ type DestinationWithBatch struct

+ DefaultBatchDelay time.Duration

+ DefaultBatchSize int

+ func (d DestinationWithBatch) Wrap(impl Destination) Destination

+ type DestinationWithRateLimit struct

+ DefaultBurst int

+ DefaultRatePerSecond float64

+ func (d DestinationWithRateLimit) Wrap(impl Destination) Destination

+ type GenerateDataType int

+ const GenerateMixedData

+ const GenerateRawData

+ const GenerateStructuredData

+ type Metadata map[string]string

+ func (m Metadata) GetConduitPluginName() (string, error)

+ func (m Metadata) GetConduitPluginVersion() (string, error)

+ func (m Metadata) GetCreatedAt() (time.Time, error)

+ func (m Metadata) GetOpenCDCVersion() (string, error)

+ func (m Metadata) GetReadAt() (time.Time, error)

+ func (m Metadata) SetConduitPluginName(name string)

+ func (m Metadata) SetConduitPluginVersion(version string)

+ func (m Metadata) SetCreatedAt(createdAt time.Time)

+ func (m Metadata) SetOpenCDCVersion()

+ func (m Metadata) SetReadAt(createdAt time.Time)

+ type Operation int

+ const OperationCreate

+ const OperationDelete

+ const OperationSnapshot

+ const OperationUpdate

+ func (i *Operation) UnmarshalText(b []byte) error

+ func (i Operation) MarshalText() ([]byte, error)

+ func (i Operation) String() string

type Record

+ Operation Operation

+ func (r Record) Bytes() []byte

type Source

+ func SourceWithMiddleware(d Source, middleware ...SourceMiddleware) Source

+ Parameters func() map[string]Parameter

+ type SourceMiddleware interface

+ Wrap func(Source) Source

+ func DefaultSourceMiddleware() []SourceMiddleware

+ type SourceUtil struct

+ func (SourceUtil) NewRecordCreate(position Position, metadata Metadata, key Data, payload Data) Record

+ func (SourceUtil) NewRecordDelete(position Position, metadata Metadata, key Data) Record

+ func (SourceUtil) NewRecordSnapshot(position Position, metadata Metadata, key Data, payload Data) Record

+ func (SourceUtil) NewRecordUpdate(position Position, metadata Metadata, key Data, payloadBefore Data, ...) Record

type UnimplementedDestination

+ func (UnimplementedDestination) Parameters() map[string]Parameter

type UnimplementedSource

+ func (UnimplementedSource) Parameters() map[string]Parameter

Apr 5, 2022

Changes in this version

+ var ErrBackoffRetry = errors.New("backoff retry")

+ var ErrUnimplemented = errors.New("action not implemented")

+ func Logger(ctx context.Context) *zerolog.Logger

+ func NewDestinationPlugin(impl Destination) cpluginv1.DestinationPlugin

+ func NewSourcePlugin(impl Source) cpluginv1.SourcePlugin

+ func NewSpecifierPlugin(specs Specification) cpluginv1.SpecifierPlugin

+ func Serve(specFactory func() Specification, sourceFactory func() Source, ...)

+ type AckFunc func(error) error

+ type Data interface

+ Bytes func() []byte

+ type Destination interface

+ Configure func(context.Context, map[string]string) error

+ Flush func(context.Context) error

+ Open func(context.Context) error

+ Teardown func(context.Context) error

+ Write func(context.Context, Record) error

+ WriteAsync func(context.Context, Record, AckFunc) error

+ type MockSource struct

+ func NewMockSource(ctrl *gomock.Controller) *MockSource

+ func (m *MockSource) Ack(arg0 context.Context, arg1 Position) error

+ func (m *MockSource) Configure(arg0 context.Context, arg1 map[string]string) error

+ func (m *MockSource) EXPECT() *MockSourceMockRecorder

+ func (m *MockSource) Open(arg0 context.Context, arg1 Position) error

+ func (m *MockSource) Read(arg0 context.Context) (Record, error)

+ func (m *MockSource) Teardown(arg0 context.Context) error

+ type MockSourceMockRecorder struct

+ func (mr *MockSourceMockRecorder) Ack(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockSourceMockRecorder) Configure(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockSourceMockRecorder) Open(arg0, arg1 interface{}) *gomock.Call

+ func (mr *MockSourceMockRecorder) Read(arg0 interface{}) *gomock.Call

+ func (mr *MockSourceMockRecorder) Teardown(arg0 interface{}) *gomock.Call

+ type Parameter struct

+ Default string

+ Description string

+ Required bool

+ type Position []byte

+ type RawData []byte

+ func (d RawData) Bytes() []byte

+ type Record struct

+ CreatedAt time.Time

+ Key Data

+ Metadata map[string]string

+ Payload Data

+ Position Position

+ type Source interface

+ Ack func(context.Context, Position) error

+ Configure func(context.Context, map[string]string) error

+ Open func(context.Context, Position) error

+ Read func(context.Context) (Record, error)

+ Teardown func(context.Context) error

+ type Specification struct

+ Author string

+ Description string

+ DestinationParams map[string]Parameter

+ Name string

+ SourceParams map[string]Parameter

+ Summary string

+ Version string

+ type StructuredData map[string]interface

+ func (d StructuredData) Bytes() []byte

+ type UnimplementedDestination struct

+ func (UnimplementedDestination) Configure(context.Context, map[string]string) error

+ func (UnimplementedDestination) Flush(context.Context) error

+ func (UnimplementedDestination) Open(context.Context) error

+ func (UnimplementedDestination) Teardown(context.Context) error

+ func (UnimplementedDestination) Write(context.Context, Record) error

+ func (UnimplementedDestination) WriteAsync(context.Context, Record, AckFunc) error

+ type UnimplementedSource struct

+ func (UnimplementedSource) Ack(context.Context, Position) error

+ func (UnimplementedSource) Configure(context.Context, map[string]string) error

+ func (UnimplementedSource) Open(context.Context, Position) error

+ func (UnimplementedSource) Read(context.Context) (Record, error)

+ func (UnimplementedSource) Teardown(context.Context) error