handler

Documentation

Index

• Variables
• func Plugin(ctx context.Context, client *wkafka.Client, config interface{}) error
• func PluginWithName() (string, wkafka.PluginFunc)
• type Config
• type DLQRecord
• type Error
  • func NewError(message string, err error, code int) Error
  • func (e Error) Error() string
• type Handler
  • func New(client *wkafka.Client, opts ...Option) (*Handler, error)
  • func (h *Handler) Event(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) File() (http.Handler, error)
  • func (h *Handler) Handler() (string, http.Handler)
  • func (h *Handler) Info(w http.ResponseWriter, _ *http.Request)
  • func (h *Handler) Serve(ctx context.Context) error
  • func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SkipDLQ(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SkipSet(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) SkipUpdate(w http.ResponseWriter, r *http.Request)
  • func (h *Handler) StartPubSub(ctx context.Context) error
  • func (h *Handler) TriggerInfo()
  • func (h *Handler) UI() (http.Handler, error)
• type Header
• type InfoResponse
• type Option
  • func WithAddr(addr string) Option
  • func WithPathPrefix(prefix string) Option
  • func WithPubSub(cfg *PubSubConfig) Option
• type PubSubConfig
  • func (c *PubSubConfig) New(groupID string, logger wkafka.Logger) (pubsub, error)
• type PubSubModel
• type PubSubModelPublish
• type Redis
  • func (r *Redis) Close() error
  • func (r *Redis) GetTopic() string
  • func (r *Redis) Publish(ctx context.Context, data PubSubModelPublish) error
  • func (r *Redis) Subscribe(ctx context.Context, handler func(PubSubModel) error) error
• type RedisConfig
• type RegexCacheStore
• type RegexPathStore
• type Response
• type SkipDLQRequest
• type SkipRequest

Variables

var PluginName = "handler"

Functions

func Plugin

func Plugin(ctx context.Context, client *wkafka.Client, config interface{}) error

func PluginWithName

func PluginWithName() (string, wkafka.PluginFunc)

Types

type Config

type Config struct {
	Enabled    bool          `cfg:"enabled"     json:"enabled"`
	Addr       string        `cfg:"addr"        json:"addr"`
	PathPrefix string        `cfg:"path_prefix" json:"path_prefix"`
	PubSub     *PubSubConfig `cfg:"pubsub"      json:"pubsub"`
}

type DLQRecord

type DLQRecord struct {
	Topic     string   `json:"topic"`
	Partition int32    `json:"partition"`
	Offset    int64    `json:"offset"`
	Key       []byte   `json:"key"`
	Value     []byte   `json:"value"`
	Timestamp string   `json:"timestamp"`
	Headers   []Header `json:"headers"`
}

type Error

type Error struct {
	Msg  string
	Err  error
	Code int
}

func NewError

func NewError(message string, err error, code int) Error

func (Error) Error

func (e Error) Error() string

type Handler

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

@Title wkafka API @BasePath /wkafka/ @description Kafka library

func New

func New(client *wkafka.Client, opts ...Option) (*Handler, error)

NewHandler returns a http.Handler implementation.

func (*Handler) Event

func (h *Handler) Event(w http.ResponseWriter, r *http.Request)

func (*Handler) File

func (h *Handler) File() (http.Handler, error)

func (*Handler) Handler

func (h *Handler) Handler() (string, http.Handler)

func (*Handler) Info

func (h *Handler) Info(w http.ResponseWriter, _ *http.Request)

Info returns the current information of the client. @Summary Get the current information of the client. @Tags wkafka @Success 200 {object} InfoResponse @Router /v1/info [GET]

func (*Handler) Serve

func (h *Handler) Serve(ctx context.Context) error

func (*Handler) ServeHTTP

func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request)

func (*Handler) SkipDLQ

func (h *Handler) SkipDLQ(w http.ResponseWriter, r *http.Request)

@Summary Skip on DLQ topic. @Tags wkafka @Accept json @Produce json @Param skip body SkipDLQRequest true "skip" @Success 200 {object} Response @Router /v1/skip-dlq [PATCH]

func (*Handler) SkipSet

func (h *Handler) SkipSet(w http.ResponseWriter, r *http.Request)

@Summary Set the skip. @Tags wkafka @Accept json @Produce json @Param skip body SkipRequest true "skip" @Success 200 {object} Response @Router /v1/skip [PUT]

func (*Handler) SkipUpdate

func (h *Handler) SkipUpdate(w http.ResponseWriter, r *http.Request)

@Summary Patch the skip. @Tags wkafka @Accept json @Produce json @Param skip body SkipRequest true "skip" @Success 200 {object} Response @Router /v1/skip [PATCH]

func (*Handler) StartPubSub

func (h *Handler) StartPubSub(ctx context.Context) error

func (*Handler) TriggerInfo

func (h *Handler) TriggerInfo()

func (*Handler) UI

func (h *Handler) UI() (http.Handler, error)

type Header

type Header struct {
	Key   string `json:"key"`
	Value []byte `json:"value"`
}

type InfoResponse

type InfoResponse struct {
	DLQTopics []string       `json:"dlq_topics"`
	Topics    []string       `json:"topics"`
	Skip      wkafka.SkipMap `json:"skip"`
	DLQRecord *DLQRecord     `json:"dlq_record"`
}

type Option

type Option func(*option)

func WithAddr

func WithAddr(addr string) Option

WithAddr to set address for handler.

• Only for Serve method, default is ":17070".

func WithPathPrefix

func WithPathPrefix(prefix string) Option

WithPathPrefix to set prefix path for handler.

func WithPubSub

func WithPubSub(cfg *PubSubConfig) Option

WithPubSub to set pubsub configuration.

type PubSubConfig

type PubSubConfig struct {
	Prefix string `cfg:"prefix" json:"prefix"`

	Redis *RedisConfig `cfg:"redis" json:"redis"`
}

func (*PubSubConfig) New

func (c *PubSubConfig) New(groupID string, logger wkafka.Logger) (pubsub, error)

type PubSubModel

type PubSubModel struct {
	Type  string          `json:"type"`
	Value json.RawMessage `json:"value"`
}

type PubSubModelPublish

type PubSubModelPublish struct {
	Type  string      `json:"type"`
	Value interface{} `json:"value"`
}

type Redis

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

func (*Redis) Close

func (r *Redis) Close() error

func (*Redis) GetTopic

func (r *Redis) GetTopic() string

func (*Redis) Publish

func (r *Redis) Publish(ctx context.Context, data PubSubModelPublish) error

func (*Redis) Subscribe

func (r *Redis) Subscribe(ctx context.Context, handler func(PubSubModel) error) error

type RedisConfig

type RedisConfig struct {
	Address  string           `cfg:"address"  json:"address"`
	Username string           `cfg:"username" json:"username"`
	Password string           `cfg:"password" json:"password"`
	TLS      wkafka.TLSConfig `cfg:"tls"      json:"tls"`
}

type RegexCacheStore

type RegexCacheStore struct {
	Regex        string `cfg:"regex"`
	CacheControl string `cfg:"cache_control"`
	// contains filtered or unexported fields
}

type RegexPathStore

type RegexPathStore struct {
	Regex       string `cfg:"regex"`
	Replacement string `cfg:"replacement"`
	// contains filtered or unexported fields
}

type Response

type Response struct {
	Message string `json:"message"`
}

type SkipDLQRequest

type SkipDLQRequest map[int32]wkafka.OffsetConfig

type SkipRequest

type SkipRequest map[string]map[int32]wkafka.OffsetConfig