Method ExpectConsumePartition

mocks/consumer.go:211–239

ExpectConsumePartition will register a topic/partition, so you can set expectations on it. The registered PartitionConsumer will be returned, so you can set expectations on it using method chaining. Once a topic/partition is registered, you are expected to start consuming it using ConsumePartition.

(topic string, partition int32, offset int64)

// and be fully consumed once the mock consumer is closed if ExpectMessagesDrainedOnClose or
// ExpectErrorsDrainedOnClose have been called.
func (c *Consumer) ExpectConsumePartition(topic string, partition int32, offset int64) *PartitionConsumer {
	c.l.Lock()
	defer c.l.Unlock()

	if c.partitionConsumers[topic] == nil {
		c.partitionConsumers[topic] = make(map[int32]*PartitionConsumer)
	}

	if c.partitionConsumers[topic][partition] == nil {
		highWatermarkOffset := offset
		if offset == sarama.OffsetOldest {
			highWatermarkOffset = 0
		}

		consumer := &PartitionConsumer{
			t:                  c.t,
			topic:              topic,
			partition:          partition,
			offset:             offset,
			messages:           make(chan *sarama.ConsumerMessage, c.config.ChannelBufferSize),
			suppressedMessages: make(chan *sarama.ConsumerMessage, c.config.ChannelBufferSize),
			errors:             make(chan *sarama.ConsumerError, c.config.ChannelBufferSize),
		}
		consumer.highWaterMarkOffset.Store(highWatermarkOffset)
		c.partitionConsumers[topic][partition] = consumer
	}

	return c.partitionConsumers[topic][partition]
}

///////////////////////////////////////////////////
// PartitionConsumer mock type