kafkatest

Documentation

Index

• Variables
• func AppNameAndArgs(cmd string) (appName string, args []string)
• func CreateTopicWithConfig(ctx context.Context, topic kafka.TopicConfig) error
• func DeleteTopic(ctx context.Context, topic string) error
• func GetOutput(cmd *exec.Cmd, wantErr bool) (io.ReadCloser, error)
• func MinifyString(str string) string
• func ReadChanMessages(ctx context.Context, c <-chan consumer.Result) []consumer.Result
• func SendMessage(t *testing.T, brokers []string, topic string, key, message []byte)
• func SendMessageToPartition(t *testing.T, brokers []string, topic string, partition int, ...)
• func SendMultipleMessagesToPartition(t *testing.T, brokers []string, topic string, msgs map[int]string)
• func StreamToRead(wantErr bool, stdout, stderr io.ReadCloser) io.ReadCloser
• type Command
  • func (c *Command) Execute(ctx context.Context)
  • func (c *Command) GetOutput() string

Variables

var DefaultTimeout = 3 * time.Second

var KafkaRawClient = kafka.Client{
	Addr:      kafka.TCP("localhost:9093"),
	Timeout:   DefaultTimeout,
	Transport: nil,
}

var LocalBroker = "localhost:9093"

Functions

func AppNameAndArgs

func AppNameAndArgs(cmd string) (appName string, args []string)

func CreateTopicWithConfig

func CreateTopicWithConfig(ctx context.Context, topic kafka.TopicConfig) error

func DeleteTopic

func DeleteTopic(ctx context.Context, topic string) error

func GetOutput

func GetOutput(cmd *exec.Cmd, wantErr bool) (io.ReadCloser, error)

func MinifyString

func MinifyString(str string) string

func ReadChanMessages

func ReadChanMessages(ctx context.Context, c <-chan consumer.Result) []consumer.Result

func SendMessage

func SendMessage(t *testing.T, brokers []string, topic string, key, message []byte)

func SendMessageToPartition

func SendMessageToPartition(t *testing.T, brokers []string, topic string, partition int, key, message []byte)

func SendMultipleMessagesToPartition

func SendMultipleMessagesToPartition(t *testing.T, brokers []string, topic string, msgs map[int]string)

func StreamToRead

func StreamToRead(wantErr bool, stdout, stderr io.ReadCloser) io.ReadCloser

Types

type Command

type Command struct {
	T *testing.T

	Cmd     string
	WantErr bool
	// contains filtered or unexported fields
}

func (*Command) Execute

func (c *Command) Execute(ctx context.Context)

func (*Command) GetOutput

func (c *Command) GetOutput() string