saramatestkit

README

Sarama Test Kit

Sarama Monkey patching collection to help to test your code

NewClient

func MyFunction() {
  // create dummy client
  dummy := saramatestkit.NewClient()
  dummy.TopicsFunc = func() ([]string, error) {
  	return []string{"topic-01", "topic-02"}, nil 
  }
  
  // monkey patch 
  patch := saramatestkit.PatchNewClient(dummy, nil)
  defer patch.Unpatch()

  // business as usual
  client, err := sarama.NewClient(nil, nil)
  if err != nil {
    log.Fatal(log.Error())
  }

  topics, err := client.Topics()
  if err != nil {
    log.Fatal(log.Error())
  }
  
  fmt.Println(topics) // {"topic-01", "topic-02"}
	
}

NewSyncProducer

func MyFunction(){
  var topic string

  // mocks.NewSyncProducer() is good but sometimes is not enough.
  // saramatestkit.NewSyncProducer() is a alternative for producer mock
  dummy := saramatestkit.NewSyncProducer()
  dummy.SendMessageFunc = func(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
  	topic = msg.Topic
  	return
  }
  
  // monkey patch
  patch := PatchNewSyncProducer(dummy, err)
  defer patch.Unpatch()
  
  // business as usual
  sarama.SendMessage(&sarama.ProducerMessage{
    Topic: "some-topic",
    Value: sarama.ByteEncoder([]byte("some-message")),
  })
  
  fmt.Println(topic) // "some-topic"
  
}

Documentation

Index

• func PatchNewClient(client sarama.Client, err error) *monkey.PatchGuard
• func PatchNewSyncProducer(producer sarama.SyncProducer, err error) *monkey.PatchGuard
• type Client
  • func NewClient() *Client
  • func (c *Client) Brokers() []*sarama.Broker
  • func (c *Client) Close() error
  • func (c *Client) Closed() bool
  • func (c *Client) Config() *sarama.Config
  • func (c *Client) Controller() (*sarama.Broker, error)
  • func (c *Client) Coordinator(consumerGroup string) (*sarama.Broker, error)
  • func (c *Client) GetOffset(topic string, partitionID int32, time int64) (int64, error)
  • func (c *Client) InSyncReplicas(topic string, partitionID int32) ([]int32, error)
  • func (c *Client) Leader(topic string, partitionID int32) (*sarama.Broker, error)
  • func (c *Client) Partitions(topic string) ([]int32, error)
  • func (c *Client) RefreshCoordinator(consumerGroup string) error
  • func (c *Client) RefreshMetadata(topics ...string) error
  • func (c *Client) Replicas(topic string, partitionID int32) ([]int32, error)
  • func (c *Client) Topics() ([]string, error)
  • func (c *Client) WritablePartitions(topic string) ([]int32, error)
• type SyncProducer
  • func NewSyncProducer() *SyncProducer
  • func (p SyncProducer) Close() error
  • func (p SyncProducer) SendMessage(msg *sarama.ProducerMessage) (int32, int64, error)
  • func (p SyncProducer) SendMessages(msgs []*sarama.ProducerMessage) error

Functions

func PatchNewClient

func PatchNewClient(client sarama.Client, err error) *monkey.PatchGuard

func PatchNewSyncProducer

func PatchNewSyncProducer(producer sarama.SyncProducer, err error) *monkey.PatchGuard

Types

type Client

type Client struct {
	ConfigFunc             func() *sarama.Config
	ControllerFunc         func() (*sarama.Broker, error)
	BrokersFunc            func() []*sarama.Broker
	TopicsFunc             func() ([]string, error)
	PartitionsFunc         func(topic string) ([]int32, error)
	WritablePartitionsFunc func(topic string) ([]int32, error)
	LeaderFunc             func(topic string, partitionID int32) (*sarama.Broker, error)
	ReplicasFunc           func(topic string, partitionID int32) ([]int32, error)
	InSyncReplicasFunc     func(topic string, partitionID int32) ([]int32, error)
	RefreshMetadataFunc    func(topics ...string) error
	GetOffsetFunc          func(topic string, partitionID int32, time int64) (int64, error)
	CoordinatorFunc        func(consumerGroup string) (*sarama.Broker, error)
	RefreshCoordinatorFunc func(consumerGroup string) error
	CloseFunc              func() error
	ClosedFunc             func() bool
}

func NewClient

func NewClient() *Client

func (*Client) Brokers

func (c *Client) Brokers() []*sarama.Broker

func (*Client) Close

func (c *Client) Close() error

func (*Client) Closed

func (c *Client) Closed() bool

func (*Client) Config

func (c *Client) Config() *sarama.Config

func (*Client) Controller

func (c *Client) Controller() (*sarama.Broker, error)

func (*Client) Coordinator

func (c *Client) Coordinator(consumerGroup string) (*sarama.Broker, error)

func (*Client) GetOffset

func (c *Client) GetOffset(topic string, partitionID int32, time int64) (int64, error)

func (*Client) InSyncReplicas

func (c *Client) InSyncReplicas(topic string, partitionID int32) ([]int32, error)

func (*Client) Leader

func (c *Client) Leader(topic string, partitionID int32) (*sarama.Broker, error)

func (*Client) Partitions

func (c *Client) Partitions(topic string) ([]int32, error)

func (*Client) RefreshCoordinator

func (c *Client) RefreshCoordinator(consumerGroup string) error

func (*Client) RefreshMetadata

func (c *Client) RefreshMetadata(topics ...string) error

func (*Client) Replicas

func (c *Client) Replicas(topic string, partitionID int32) ([]int32, error)

func (*Client) Topics

func (c *Client) Topics() ([]string, error)

func (*Client) WritablePartitions

func (c *Client) WritablePartitions(topic string) ([]int32, error)

type SyncProducer

type SyncProducer struct {
	SendMessageFunc  func(msg *sarama.ProducerMessage) (partition int32, offset int64, err error)
	SendMessagesFunc func(msgs []*sarama.ProducerMessage) error
	CloseFunc        func() error
}

func NewSyncProducer

func NewSyncProducer() *SyncProducer

func (SyncProducer) Close

func (p SyncProducer) Close() error

func (SyncProducer) SendMessage

func (p SyncProducer) SendMessage(msg *sarama.ProducerMessage) (int32, int64, error)

func (SyncProducer) SendMessages

func (p SyncProducer) SendMessages(msgs []*sarama.ProducerMessage) error