kafka

Documentation

Index

• Constants
• type Config
• type Endpoint
• type Kafka
  • func (x *Kafka) AddRouter(router endpointApi.Router, params ...interface{}) (string, error)
  • func (x *Kafka) Close() error
  • func (x *Kafka) Destroy()
  • func (x *Kafka) Id() string
  • func (x *Kafka) Init(ruleConfig types.Config, configuration types.Configuration) error
  • func (x *Kafka) New() types.Node
  • func (x *Kafka) Printf(format string, v ...interface{})
  • func (x *Kafka) RemoveRouter(routerId string, params ...interface{}) error
  • func (x *Kafka) Start() error
  • func (x *Kafka) Type() string
• type RequestMessage
  • func (r *RequestMessage) Body() []byte
  • func (r *RequestMessage) From() string
  • func (r *RequestMessage) GetError() error
  • func (r *RequestMessage) GetMsg() *types.RuleMsg
  • func (r *RequestMessage) GetParam(key string) string
  • func (r *RequestMessage) Headers() textproto.MIMEHeader
  • func (r *RequestMessage) SetBody(body []byte)
  • func (r *RequestMessage) SetError(err error)
  • func (r *RequestMessage) SetMsg(msg *types.RuleMsg)
  • func (r *RequestMessage) SetStatusCode(statusCode int)
• type ResponseMessage
  • func (r *ResponseMessage) Body() []byte
  • func (r *ResponseMessage) From() string
  • func (r *ResponseMessage) GetError() error
  • func (r *ResponseMessage) GetMsg() *types.RuleMsg
  • func (r *ResponseMessage) GetParam(key string) string
  • func (r *ResponseMessage) Headers() textproto.MIMEHeader
  • func (r *ResponseMessage) SetBody(body []byte)
  • func (r *ResponseMessage) SetError(err error)
  • func (r *ResponseMessage) SetMsg(msg *types.RuleMsg)
  • func (r *ResponseMessage) SetStatusCode(statusCode int)

Constants

const (
	//Topic 消息主题
	Topic = "topic"
	//Key 消息key
	Key = "key"
	//Partition 消费分区
	Partition = "partition"
)

const (
	// KeyResponseTopic 响应主题metadataKey
	KeyResponseTopic = "responseTopic"
	// KeyResponseKey 响应key metadataKey
	KeyResponseKey = "key"
	// KeyResponsePartition 响应 消费分区metadataKey
	KeyResponsePartition = "partition"
)

const Type = types.EndpointTypePrefix + "kafka"

Type 组件类型

Types

type Config

type Config struct {
	// kafka服务器地址列表，多个与逗号隔开
	Server string
	// GroupId 消费者组Id
	GroupId string
}

type Endpoint

type Endpoint = Kafka

Endpoint 别名

type Kafka

type Kafka struct {
	impl.BaseEndpoint
	RuleConfig types.Config
	//Config 配置
	Config Config
	// contains filtered or unexported fields
}

Kafka Kafka 接收端端点

func (*Kafka) AddRouter

func (x *Kafka) AddRouter(router endpointApi.Router, params ...interface{}) (string, error)

func (*Kafka) Close

func (x *Kafka) Close() error

func (*Kafka) Destroy

func (x *Kafka) Destroy()

Destroy 销毁

func (*Kafka) Id

func (x *Kafka) Id() string

func (*Kafka) Init

func (x *Kafka) Init(ruleConfig types.Config, configuration types.Configuration) error

Init 初始化

func (*Kafka) New

func (x *Kafka) New() types.Node

func (*Kafka) Printf

func (x *Kafka) Printf(format string, v ...interface{})

func (*Kafka) RemoveRouter

func (x *Kafka) RemoveRouter(routerId string, params ...interface{}) error

func (*Kafka) Start

func (x *Kafka) Start() error

func (*Kafka) Type

func (x *Kafka) Type() string

Type 组件类型

type RequestMessage

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

RequestMessage http请求消息

func (*RequestMessage) Body

func (r *RequestMessage) Body() []byte

func (*RequestMessage) From

func (r *RequestMessage) From() string

func (*RequestMessage) GetError

func (r *RequestMessage) GetError() error

func (*RequestMessage) GetMsg

func (r *RequestMessage) GetMsg() *types.RuleMsg

func (*RequestMessage) GetParam

func (r *RequestMessage) GetParam(key string) string

func (*RequestMessage) Headers

func (r *RequestMessage) Headers() textproto.MIMEHeader

func (*RequestMessage) SetBody

func (r *RequestMessage) SetBody(body []byte)

func (*RequestMessage) SetError

func (r *RequestMessage) SetError(err error)

func (*RequestMessage) SetMsg

func (r *RequestMessage) SetMsg(msg *types.RuleMsg)

func (*RequestMessage) SetStatusCode

func (r *RequestMessage) SetStatusCode(statusCode int)

type ResponseMessage

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

ResponseMessage http响应消息

func (*ResponseMessage) Body

func (r *ResponseMessage) Body() []byte

func (*ResponseMessage) From

func (r *ResponseMessage) From() string

func (*ResponseMessage) GetError

func (r *ResponseMessage) GetError() error

func (*ResponseMessage) GetMsg

func (r *ResponseMessage) GetMsg() *types.RuleMsg

func (*ResponseMessage) GetParam

func (r *ResponseMessage) GetParam(key string) string

func (*ResponseMessage) Headers

func (r *ResponseMessage) Headers() textproto.MIMEHeader

func (*ResponseMessage) SetBody

func (r *ResponseMessage) SetBody(body []byte)

func (*ResponseMessage) SetError

func (r *ResponseMessage) SetError(err error)

func (*ResponseMessage) SetMsg

func (r *ResponseMessage) SetMsg(msg *types.RuleMsg)

func (*ResponseMessage) SetStatusCode

func (r *ResponseMessage) SetStatusCode(statusCode int)