client.go

package cluster

import (
	"errors"
	"sync/atomic"

	"github.com/Shopify/sarama"
)

var errClientInUse = errors.New("cluster: client is already used by another consumer")

// Client is a group client
type Client struct {
	sarama.Client
	config Config

	inUse uint32
}

// NewClient creates a new client instance
func NewClient(addrs []string, config *Config) (*Client, error) {
	if config == nil {
		config = NewConfig()
	}

	if err := config.Validate(); err != nil {
		return nil, err
	}

	client, err := sarama.NewClient(addrs, &config.Config)
	if err != nil {
		return nil, err
	}

	return &Client{Client: client, config: *config}, nil
}

// ClusterConfig returns the cluster configuration.
func (c *Client) ClusterConfig() *Config {
	cfg := c.config
	return &cfg
}

func (c *Client) claim() bool {
	return atomic.CompareAndSwapUint32(&c.inUse, 0, 1)
}

func (c *Client) release() {
	atomic.CompareAndSwapUint32(&c.inUse, 1, 0)
}