Function newConsumer

consumer.go:129–144

(client Client)

}

func newConsumer(client Client) (Consumer, error) {
	// Check that we are not dealing with a closed Client before processing any other arguments
	if client.Closed() {
		return nil, ErrClosedClient
	}

	c := &consumer{
		client:          client,
		conf:            client.Config(),
		children:        make(map[string]map[int32]*partitionConsumer),
		brokerConsumers: make(map[*Broker]*brokerConsumer),
		metricRegistry:  newCleanupRegistry(client.Config().MetricRegistry),
	}

	return c, nil
}

func (c *consumer) Close() error {
	c.metricRegistry.UnregisterAll()