contube

package

Go to main page

Versions in this module

v0

v0.5.0

Jun 13, 2024

Changes in this version

+ const IncludeMetadata

+ var ErrSourceTubeNotImplemented = errors.Wrap(ErrTubeNotImplemented, "source tube not implemented")

+ var ErrTubeNotImplemented = errors.New("tube not implemented")

type ConfigMap

+ func ToConfigMap(v any) (ConfigMap, error)

+ type HttpHandler func(w http.ResponseWriter, r *http.Request, payload []byte) Record

type HttpTubeFactory

+ func NewHttpTubeFactoryWithIntercept(ctx context.Context, handler HttpHandler) *HttpTubeFactory

type Record

+ func DefaultHttpHandler(_ http.ResponseWriter, _ *http.Request, payload []byte) Record

+ GetSchema func() string

type RecordImpl

+ func NewSchemaRecordImpl(payload []byte, schema string, ackFunc func()) *RecordImpl

+ func (e *RecordImpl) GetSchema() string

v0.4.0

May 7, 2024

Changes in this version

type ConfigMap

+ func (c ConfigMap) ToConfigStruct(v any) error

v0.3.0

Mar 8, 2024

Changes in this version

+ const EndpointKey

+ const PulsarURLKey

+ const SubNameKey

+ const TopicKey

+ const TopicListKey

+ var ErrEndpointClosed = errors.New("endpoint closed")

+ var ErrEndpointNotFound = errors.New("endpoint not found")

+ var ErrSinkTubeNotImplemented = errors.New("sink tube not implemented")

+ var ErrorEndpointAlreadyExists = errors.New("endpoint already exists")

+ type ConfigMap map[string]interface

+ func MergeConfig(configs ...ConfigMap) ConfigMap

+ type EndpointHandler func(ctx context.Context, endpoint string, payload []byte) error

+ type HttpTubeFactory struct

+ func NewHttpTubeFactory(ctx context.Context) *HttpTubeFactory

+ func (f *HttpTubeFactory) GetHandleFunc(getEndpoint func(r *http.Request) (string, error), logger *slog.Logger) func(http.ResponseWriter, *http.Request)

+ func (f *HttpTubeFactory) Handle(ctx context.Context, endpoint string, payload []byte) error

+ func (f *HttpTubeFactory) NewSinkTube(_ context.Context, _ ConfigMap) (chan<- Record, error)

+ func (f *HttpTubeFactory) NewSourceTube(ctx context.Context, config ConfigMap) (<-chan Record, error)

+ type MemoryQueueFactory struct

+ func (f *MemoryQueueFactory) NewSinkTube(ctx context.Context, configMap ConfigMap) (chan<- Record, error)

+ func (f *MemoryQueueFactory) NewSourceTube(ctx context.Context, configMap ConfigMap) (<-chan Record, error)

+ type PulsarEventQueueFactory struct

+ func (f *PulsarEventQueueFactory) NewSinkTube(ctx context.Context, configMap ConfigMap) (chan<- Record, error)

+ func (f *PulsarEventQueueFactory) NewSourceTube(ctx context.Context, configMap ConfigMap) (<-chan Record, error)

+ type PulsarTubeFactoryConfig struct

+ PulsarURL string

+ func NewPulsarTubeFactoryConfig(configMap ConfigMap) *PulsarTubeFactoryConfig

+ func (c *PulsarTubeFactoryConfig) ToConfigMap() ConfigMap

+ type Record interface

+ Commit func()

+ GetPayload func() []byte

+ type RecordImpl struct

+ func NewRecordImpl(payload []byte, ackFunc func()) *RecordImpl

+ func (e *RecordImpl) Commit()

+ func (e *RecordImpl) GetPayload() []byte

+ type SinkQueueConfig struct

+ Topic string

+ func NewSinkQueueConfig(config ConfigMap) *SinkQueueConfig

+ func (c *SinkQueueConfig) ToConfigMap() ConfigMap

+ type SinkTubeFactory interface

+ NewSinkTube func(ctx context.Context, config ConfigMap) (chan<- Record, error)

+ type SourceQueueConfig struct

+ SubName string

+ Topics []string

+ func NewSourceQueueConfig(config ConfigMap) *SourceQueueConfig

+ func (c *SourceQueueConfig) ToConfigMap() ConfigMap

+ type SourceTubeFactory interface

+ NewSourceTube func(ctx context.Context, config ConfigMap) (<-chan Record, error)

+ type TubeFactory interface

+ func NewMemoryQueueFactory(ctx context.Context) TubeFactory

+ func NewPulsarEventQueueFactory(ctx context.Context, configMap ConfigMap) (TubeFactory, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL