contube

Documentation

Index

• Constants
• Variables
• type ConfigMap
  • func MergeConfig(configs ...ConfigMap) ConfigMap
  • func ToConfigMap(v any) (ConfigMap, error)
  • func (c ConfigMap) ToConfigStruct(v any) error
• type EndpointHandler
• type HttpHandler
• type HttpTubeFactory
  • func NewHttpTubeFactory(ctx context.Context) *HttpTubeFactory
  • func NewHttpTubeFactoryWithIntercept(ctx context.Context, handler HttpHandler) *HttpTubeFactory
  • func (f *HttpTubeFactory) GetHandleFunc(getEndpoint func(r *http.Request) (string, error), logger *common.Logger) func(http.ResponseWriter, *http.Request)
  • func (f *HttpTubeFactory) Handle(ctx context.Context, endpoint string, record Record) 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
  • 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
  • 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
  • func NewPulsarTubeFactoryConfig(configMap ConfigMap) (*PulsarTubeFactoryConfig, error)
  • func (c *PulsarTubeFactoryConfig) ToConfigMap() ConfigMap
• type Record
  • func DefaultHttpHandler(_ http.ResponseWriter, _ *http.Request, payload []byte) Record
• type RecordImpl
  • func NewRecordImpl(payload []byte, ackFunc func()) *RecordImpl
  • func NewSchemaRecordImpl(payload []byte, schema string, ackFunc func()) *RecordImpl
  • func (e *RecordImpl) Commit()
  • func (e *RecordImpl) GetPayload() []byte
  • func (e *RecordImpl) GetSchema() string
• type SinkQueueConfig
  • func (c *SinkQueueConfig) ToConfigMap() ConfigMap
• type SinkTubeFactory
• type SourceQueueConfig
  • func (c *SourceQueueConfig) ToConfigMap() ConfigMap
• type SourceTubeFactory
• type TubeFactory
  • func NewMemoryQueueFactory(ctx context.Context) TubeFactory
  • func NewPulsarEventQueueFactory(ctx context.Context, configMap ConfigMap) (TubeFactory, error)

Constants

const (
	EndpointKey     = "endpoint"
	IncludeMetadata = "includeMetadata"
)

const (
	PulsarURLKey = "pulsar_url"
)

Variables

var (
	ErrTubeNotImplemented       = errors.New("tube not implemented")
	ErrSinkTubeNotImplemented   = errors.Wrap(ErrTubeNotImplemented, "sink tube not implemented")
	ErrSourceTubeNotImplemented = errors.Wrap(ErrTubeNotImplemented, "source tube not implemented")
)

var (
	ErrEndpointNotFound        = errors.New("endpoint not found")
	ErrEndpointClosed          = errors.New("endpoint closed")
	ErrorEndpointAlreadyExists = errors.New("endpoint already exists")
)

Types

type ConfigMap

type ConfigMap map[string]interface{}

func MergeConfig

func MergeConfig(configs ...ConfigMap) ConfigMap

MergeConfig merges multiple ConfigMap into one

func ToConfigMap

func ToConfigMap(v any) (ConfigMap, error)

func (ConfigMap) ToConfigStruct

func (c ConfigMap) ToConfigStruct(v any) error

type EndpointHandler

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

type HttpHandler

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

type HttpTubeFactory

type HttpTubeFactory struct {
	TubeFactory
	// contains filtered or unexported fields
}

func NewHttpTubeFactory

func NewHttpTubeFactory(ctx context.Context) *HttpTubeFactory

func NewHttpTubeFactoryWithIntercept

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

func (*HttpTubeFactory) GetHandleFunc

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

func (*HttpTubeFactory) Handle

func (f *HttpTubeFactory) Handle(ctx context.Context, endpoint string, record Record) error

func (*HttpTubeFactory) NewSinkTube

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

func (*HttpTubeFactory) NewSourceTube

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

type MemoryQueueFactory

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

func (*MemoryQueueFactory) NewSinkTube

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

func (*MemoryQueueFactory) NewSourceTube

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

type PulsarEventQueueFactory

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

func (*PulsarEventQueueFactory) NewSinkTube

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

func (*PulsarEventQueueFactory) NewSourceTube

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

type PulsarTubeFactoryConfig

type PulsarTubeFactoryConfig struct {
	PulsarURL string
}

func NewPulsarTubeFactoryConfig

func NewPulsarTubeFactoryConfig(configMap ConfigMap) (*PulsarTubeFactoryConfig, error)

func (*PulsarTubeFactoryConfig) ToConfigMap

func (c *PulsarTubeFactoryConfig) ToConfigMap() ConfigMap

type Record

type Record interface {
	GetPayload() []byte
	GetSchema() string
	Commit()
}

func DefaultHttpHandler

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

type RecordImpl

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

func NewRecordImpl

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

func NewSchemaRecordImpl

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

func (*RecordImpl) Commit

func (e *RecordImpl) Commit()

func (*RecordImpl) GetPayload

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

func (*RecordImpl) GetSchema

func (e *RecordImpl) GetSchema() string

type SinkQueueConfig

type SinkQueueConfig struct {
	Topic string `json:"output" validate:"required"`
}

func (*SinkQueueConfig) ToConfigMap

func (c *SinkQueueConfig) ToConfigMap() ConfigMap

type SinkTubeFactory

type SinkTubeFactory interface {
	// NewSinkTube returns a new channel that can be used to sink events
	// The event.Commit() would be invoked after the event is sunk successfully
	// The caller should close the channel when it is done
	NewSinkTube(ctx context.Context, config ConfigMap) (chan<- Record, error)
}

type SourceQueueConfig

type SourceQueueConfig struct {
	Topics  []string `json:"inputs" validate:"required"`
	SubName string   `json:"subscription-name" validate:"required"`
}

func (*SourceQueueConfig) ToConfigMap

func (c *SourceQueueConfig) ToConfigMap() ConfigMap

type SourceTubeFactory

type SourceTubeFactory interface {
	// NewSourceTube returns a new channel that can be used to receive events
	// The channel would be closed when the context is done
	NewSourceTube(ctx context.Context, config ConfigMap) (<-chan Record, error)
}

type TubeFactory

type TubeFactory interface {
	SourceTubeFactory
	SinkTubeFactory
}

func NewMemoryQueueFactory

func NewMemoryQueueFactory(ctx context.Context) TubeFactory

func NewPulsarEventQueueFactory

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