ronantakizawa/codereview-bench huggingface dataset:
Repository: nsqio/nsq
File: nsqd/client_v2.go (Go)

Code before:
	}

	err = c.SetMsgTimeout(data.MsgTimeout)
	if err != nil {
		return err
	}

	ie := identifyEvent{
		OutputBufferTimeout: c.OutputBufferTimeout,
		HeartbeatInterval:   c.HeartbeatInterval,
		SampleRate:          c.SampleRate,
		MsgTimeout:          c.MsgTimeout,
	}

	// update the client's message pump
	select {
	case c.IdentifyEventChan <- ie:
	default:
	}

	return nil
}

func (c *clientV2) Stats() ClientStats {
	c.RLock()
	// TODO: deprecated, remove in 1.0
	name := c.ClientID

	clientID := c.ClientID
	hostname := c.Hostname
	userAgent := c.UserAgent
	var identity string
	var identityURL string
	if c.AuthState != nil {
		identity = c.AuthState.Identity
		identityURL = c.AuthState.IdentityURL
	}
	c.RUnlock()
	stats := ClientStats{
		// TODO: deprecated, remove in 1.0
		Name: name,

		Version:         "V2",
		RemoteAddress:   c.RemoteAddr().String(),
		ClientID:        clientID,
		Hostname:        hostname,
		UserAgent:       userAgent,
		State:           atomic.LoadInt32(&c.State),
		ReadyCount:      atomic.LoadInt64(&c.ReadyCount),
		InFlightCount:   atomic.LoadInt64(&c.InFlightCount),
		MessageCount:    atomic.LoadUint64(&c.MessageCount),

Review comment (suggestion):
FYI this was ultimately the most problematic case.

Previously, this was competing with network writes to this client [here](https://github.com/nsqio/nsq/blob/master/nsqd/protocol_v2.go#L147), so when that blocked, this blocked.  It's called as part of the underlying code path for `/stats` [here](https://github.com/nsqio/nsq/pull/701/files#diff-710982d0c740f14afd7d0153750a9370R140).  Prior to #700, the body of `nsqd.GetStats()` would hold an nsqd-wide lock for the entire lifetime of the function call, meaning that a single blocking client could effectively block any operations that required a top-level nsqd-wide lock, or a topic/channel-level lock for whichever the misbehaving client was subscribed to.


Code after:
		return err
	}

	err = c.SetMsgTimeout(data.MsgTimeout)
	if err != nil {
		return err
	}

	ie := identifyEvent{
		OutputBufferTimeout: c.OutputBufferTimeout,
		HeartbeatInterval:   c.HeartbeatInterval,
		SampleRate:          c.SampleRate,
		MsgTimeout:          c.MsgTimeout,
	}

	// update the client's message pump
	select {
	case c.IdentifyEventChan <- ie:
	default:
	}

	return nil
}

func (c *clientV2) Stats() ClientStats {
	c.metaLock.RLock()
	// TODO: deprecated, remove in 1.0
	name := c.ClientID

	clientID := c.ClientID
	hostname := c.Hostname
	userAgent := c.UserAgent
	var identity string
	var identityURL string
	if c.AuthState != nil {
		identity = c.AuthState.Identity
		identityURL = c.AuthState.IdentityURL
	}
	c.metaLock.RUnlock()
	stats := ClientStats{
		// TODO: deprecated, remove in 1.0
		Name: name,

		Version:         "V2",
		RemoteAddress:   c.RemoteAddr().String(),
		ClientID:        clientID,
		Hostname:        hostname,
		UserAgent:       userAgent,
		State:           atomic.LoadInt32(&c.State),
		ReadyCount:      atomic.LoadInt64(&c.ReadyCount),
		InFlightCount:   atomic.LoadInt64(&c.InFlightCount),
Repository: nsqio/nsq
File: nsqd/client_v2.go (Go)

Code before:
	}

	err = c.SetMsgTimeout(data.MsgTimeout)
	if err != nil {
		return err
	}

	ie := identifyEvent{
		OutputBufferTimeout: c.OutputBufferTimeout,
		HeartbeatInterval:   c.HeartbeatInterval,
		SampleRate:          c.SampleRate,
		MsgTimeout:          c.MsgTimeout,
	}

	// update the client's message pump
	select {
	case c.IdentifyEventChan <- ie:
	default:
	}

	return nil
}

func (c *clientV2) Stats() ClientStats {
	c.RLock()
	// TODO: deprecated, remove in 1.0
	name := c.ClientID

	clientID := c.ClientID
	hostname := c.Hostname
	userAgent := c.UserAgent
	var identity string
	var identityURL string
	if c.AuthState != nil {
		identity = c.AuthState.Identity
		identityURL = c.AuthState.IdentityURL
	}
	c.RUnlock()
	stats := ClientStats{
		// TODO: deprecated, remove in 1.0
		Name: name,

		Version:         "V2",
		RemoteAddress:   c.RemoteAddr().String(),
		ClientID:        clientID,
		Hostname:        hostname,
		UserAgent:       userAgent,
		State:           atomic.LoadInt32(&c.State),
		ReadyCount:      atomic.LoadInt64(&c.ReadyCount),
		InFlightCount:   atomic.LoadInt64(&c.InFlightCount),
		MessageCount:    atomic.LoadUint64(&c.MessageCount),

Review comment (suggestion):
P.S. this has the really nice side-effect of making `/stats` requests _really_ fast, regardless of `nsqd` load, addressing long-standing issues with `nsqadmin` timing out.


Code after:
		return err
	}

	err = c.SetMsgTimeout(data.MsgTimeout)
	if err != nil {
		return err
	}

	ie := identifyEvent{
		OutputBufferTimeout: c.OutputBufferTimeout,
		HeartbeatInterval:   c.HeartbeatInterval,
		SampleRate:          c.SampleRate,
		MsgTimeout:          c.MsgTimeout,
	}

	// update the client's message pump
	select {
	case c.IdentifyEventChan <- ie:
	default:
	}

	return nil
}

func (c *clientV2) Stats() ClientStats {
	c.metaLock.RLock()
	// TODO: deprecated, remove in 1.0
	name := c.ClientID

	clientID := c.ClientID
	hostname := c.Hostname
	userAgent := c.UserAgent
	var identity string
	var identityURL string
	if c.AuthState != nil {
		identity = c.AuthState.Identity
		identityURL = c.AuthState.IdentityURL
	}
	c.metaLock.RUnlock()
	stats := ClientStats{
		// TODO: deprecated, remove in 1.0
		Name: name,

		Version:         "V2",
		RemoteAddress:   c.RemoteAddr().String(),
		ClientID:        clientID,
		Hostname:        hostname,
		UserAgent:       userAgent,
		State:           atomic.LoadInt32(&c.State),
		ReadyCount:      atomic.LoadInt64(&c.ReadyCount),
		InFlightCount:   atomic.LoadInt64(&c.InFlightCount),
Repository: nsqio/nsq
File: apps/nsqd/nsqd.go (Go)

Code before:
		if err == nil {
			newVal := fmt.Sprintf("%v", t.Get())
			if newVal != "0" {
				cfg["tls_min_version"] = newVal
			} else {
				delete(cfg, "tls_min_version")
			}
		} else {
			log.Fatalf("ERROR: failed parsing tls min version %v", v)
		}
	}
}

func main() {
	flagSet := nsqFlagset()
	flagSet.Parse(os.Args[1:])

	rand.Seed(time.Now().UTC().UnixNano())

	if flagSet.Lookup("version").Value.(flag.Getter).Get().(bool) {
		fmt.Println(version.String("nsqd"))
		return
	}

	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)

	var cfg config
	configFile := flagSet.Lookup("config").Value.String()
	if configFile != "" {
		_, err := toml.DecodeFile(configFile, &cfg)
		if err != nil {
			log.Fatalf("ERROR: failed to load config file %s - %s", configFile, err.Error())
		}
	}
	cfg.Validate()

	opts := nsqd.NewOptions()
	options.Resolve(opts, flagSet, cfg)
	nsqd := nsqd.New(opts)

	nsqd.LoadMetadata()
	err := nsqd.PersistMetadata()
	if err != nil {
		log.Fatalf("ERROR: failed to persist metadata - %s", err.Error())
	}
	nsqd.Main()
	<-signalChan
	nsqd.Exit()
}


Review comment (suggestion):
It's fine for now, but perhaps users of `go-svc` want to specify what signals they want to handle?  Right now it's hardcoded https://github.com/judwhite/go-svc/blob/master/svc/svc_other.go#L19


Code after:
		if err == nil {
			newVal := fmt.Sprintf("%v", t.Get())
			if newVal != "0" {
				cfg["tls_min_version"] = newVal
			} else {
				delete(cfg, "tls_min_version")
			}
		} else {
			log.Fatalf("ERROR: failed parsing tls min version %v", v)
		}
	}
}

type program struct {
	nsqd *nsqd.NSQD
}

func main() {
	prg := &program{}
	if err := svc.Run(prg); err != nil {
		log.Fatal(err)
	}
}

func (p *program) Init(env svc.Environment) error {
	if env.IsWindowsService() {
		dir := filepath.Dir(os.Args[0])
		return os.Chdir(dir)
	}
	return nil
}

func (p *program) Start() error {
	flagSet := nsqFlagset()
	flagSet.Parse(os.Args[1:])

	rand.Seed(time.Now().UTC().UnixNano())

	if flagSet.Lookup("version").Value.(flag.Getter).Get().(bool) {
		fmt.Println(version.String("nsqd"))
		os.Exit(0)
	}

	var cfg config
	configFile := flagSet.Lookup("config").Value.String()
	if configFile != "" {
		_, err := toml.DecodeFile(configFile, &cfg)
		if err != nil {
			log.Fatalf("ERROR: failed to load config file %s - %s", configFile, err.Error())
		}
	}
Repository: nsqio/nsq
File: internal/http_api/api_request.go (Go)

Code before:

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

type deadlinedConn struct {
	Timeout time.Duration
	net.Conn
}

func (c *deadlinedConn) Read(b []byte) (n int, err error) {
	return c.Conn.Read(b)
}

func (c *deadlinedConn) Write(b []byte) (n int, err error) {
	return c.Conn.Write(b)
}

// A custom http.Transport with support for deadline timeouts
func NewDeadlineTransport(timeout time.Duration) *http.Transport {
	transport := &http.Transport{
		Dial: func(netw, addr string) (net.Conn, error) {
			c, err := net.DialTimeout(netw, addr, timeout)
			if err != nil {
				return nil, err
			}
			return &deadlinedConn{timeout, c}, nil
		},
	}
	return transport
}

type Client struct {
	c *http.Client
}

func NewClient(tlsConfig *tls.Config, clientTimeout time.Duration, requestTimeout time.Duration) *Client {
	transport := NewDeadlineTransport(clientTimeout)
	transport.TLSClientConfig = tlsConfig
	return &Client{
		c: &http.Client{
			Transport: transport,

Review comment (suggestion):
Now that we've removed these, we need to set `ResponseHeaderTimeout` to our configured request timeout on the `http.Transport` instantiation below on line 30. `NewDeadlineTransport` will need to take both connect timeout and request timeout.


Code after:

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
)

type deadlinedConn struct {
	Timeout time.Duration
	net.Conn
}

func (c *deadlinedConn) Read(b []byte) (n int, err error) {
	return c.Conn.Read(b)
}

func (c *deadlinedConn) Write(b []byte) (n int, err error) {
	return c.Conn.Write(b)
}

// A custom http.Transport with support for deadline timeouts
func NewDeadlineTransport(connectTimeout time.Duration, requestTimeout time.Duration) *http.Transport {
	transport := &http.Transport{
		Dial: func(netw, addr string) (net.Conn, error) {
			c, err := net.DialTimeout(netw, addr, connectTimeout)
			if err != nil {
				return nil, err
			}
			return &deadlinedConn{connectTimeout, c}, nil
		},
		ResponseHeaderTimeout: requestTimeout,
	}
	return transport
}

type Client struct {
	c *http.Client
}

func NewClient(tlsConfig *tls.Config, connectTimeout time.Duration, requestTimeout time.Duration) *Client {
	transport := NewDeadlineTransport(connectTimeout, requestTimeout)
	transport.TLSClientConfig = tlsConfig
	return &Client{
		c: &http.Client{
Repository: nsqio/nsq
File: nsqd/nsqd_test.go (Go)

Code before:

	_, _, nsqd = mustStartNSQD(opts)
	err = nsqd.LoadMetadata()
	test.Nil(t, err)

	t1, err := nsqd.GetExistingTopic("t1")
	test.Nil(t, err)
	test.NotNil(t, t1)
	c2, err := t1.GetExistingChannel("c2")
	test.Nil(t, err)
	test.NotNil(t, c2)

	nsqd.Exit()
}

func TestMetadataConflict(t *testing.T) {
	old_meta := `
	{
	  "topics": [{
	    "name": "t1", "paused": false,
	    "channels": [{"name": "c1", "paused": false}]
	  }],
	  "version": "1.0.0-alpha"
	}`
	new_meta := `
	{
	  "topics": [{
	    "name": "t2", "paused": false,
	    "channels": [{"name": "c2", "paused": false}]
	  }],
	  "version": "1.0.0-alpha"
	}`

	tmpDir, err := ioutil.TempDir("", "nsq-test-")
	if err != nil {
		panic(err)
	}
	defer os.RemoveAll(tmpDir)

	opts := NewOptions()
	opts.DataPath = tmpDir
	opts.Logger = test.NewTestLogger(t)

	err = ioutil.WriteFile(oldMetadataFile(opts), []byte(old_meta), 0700)
	if err != nil {
		panic(err)
	}
	err = ioutil.WriteFile(newMetadataFile(opts), []byte(new_meta), 0700)
	if err != nil {
		panic(err)
	}

Review comment (refactor):
`TempDir()` does the random-suffix part automatically. Docs say:

> Multiple programs calling TempDir simultaneously will not choose the same directory.

(I mention this because the `mustStartNSQD()` helper composes its own random-ish suffix for TempDir() if DataPath is unset.)

Code after:
	c2, err := t1.GetExistingChannel("c2")
	test.Nil(t, err)
	test.NotNil(t, c2)

	nsqd.Exit()
}

func TestMetadataConflict(t *testing.T) {
	old_meta := `
	{
	  "topics": [{
	    "name": "t1", "paused": false,
	    "channels": [{"name": "c1", "paused": false}]
	  }],
	  "version": "1.0.0-alpha"
	}`
	new_meta := `
	{
	  "topics": [{
	    "name": "t2", "paused": false,
	    "channels": [{"name": "c2", "paused": false}]
	  }],
	  "version": "1.0.0-alpha"
	}`

	tmpDir, err := ioutil.TempDir("", "nsq-test-")
	if err != nil {
		panic(err)
	}
	defer os.RemoveAll(tmpDir)

	opts := NewOptions()
	opts.DataPath = tmpDir
	opts.Logger = test.NewTestLogger(t)

	err = ioutil.WriteFile(oldMetadataFile(opts), []byte(old_meta), 0600)
	if err != nil {
		panic(err)
	}
	err = ioutil.WriteFile(newMetadataFile(opts), []byte(new_meta), 0600)
	if err != nil {
		panic(err)
	}

	_, _, nsqd := mustStartNSQD(opts)
	err = nsqd.LoadMetadata()
	test.NotNil(t, err)
	nsqd.Exit()
}

func TestEphemeralTopicsAndChannels(t *testing.T) {
Repository: nsqio/nsq
File: nsqadmin/nsqadmin.go (Go)

Code before:
	if opts.AllowConfigFromCIDR != "" {
		_, _, err := net.ParseCIDR(opts.AllowConfigFromCIDR)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --allow-config-from-cidr='%s' - %s", opts.AllowConfigFromCIDR, err)
			os.Exit(1)
		}
	}

	opts.BasePath = normalizeBasePath(path.Clean(opts.BasePath))

	n.logf(LOG_INFO, version.String("nsqadmin"))

	return n
}

func normalizeBasePath(p string) string {
	if len(p) == 0 {
		return "/"
	}
	// add leading slash
	if p[0] != '/' {
		p = "/" + p
	}
	// add trailing slash
	if p[len(p)-1] != '/' {
		p = p + "/"
	}
	return p
}

func (n *NSQAdmin) getOpts() *Options {
	return n.opts.Load().(*Options)
}

func (n *NSQAdmin) swapOpts(opts *Options) {
	n.opts.Store(opts)
}

func (n *NSQAdmin) RealHTTPAddr() *net.TCPAddr {
	return n.httpListener.Addr().(*net.TCPAddr)
}

func (n *NSQAdmin) handleAdminActions() {
	for action := range n.notifications {
		content, err := json.Marshal(action)
		if err != nil {
			n.logf(LOG_ERROR, "failed to serialize admin action - %s", err)
		}
		httpclient := &http.Client{
			Transport: http_api.NewDeadlineTransport(n.getOpts().HTTPClientConnectTimeout, n.getOpts().HTTPClientRequestTimeout),
		}

Review comment (suggestion):
I think the templated urls and such may look a bit better if instead `BasePath` does _not_ end in `/` and we write the `/` after it everywhere. But I understand if you disagree, and it's OK this way too.

Code after:
	if opts.AllowConfigFromCIDR != "" {
		_, _, err := net.ParseCIDR(opts.AllowConfigFromCIDR)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --allow-config-from-cidr='%s' - %s", opts.AllowConfigFromCIDR, err)
			os.Exit(1)
		}
	}

	opts.BasePath = normalizeBasePath(opts.BasePath)

	n.logf(LOG_INFO, version.String("nsqadmin"))

	return n
}

func normalizeBasePath(p string) string {
	if len(p) == 0 {
		return "/"
	}
	// add leading slash
	if p[0] != '/' {
		p = "/" + p
	}
	return path.Clean(p)
}

func (n *NSQAdmin) getOpts() *Options {
	return n.opts.Load().(*Options)
}

func (n *NSQAdmin) swapOpts(opts *Options) {
	n.opts.Store(opts)
}

func (n *NSQAdmin) RealHTTPAddr() *net.TCPAddr {
	return n.httpListener.Addr().(*net.TCPAddr)
}

func (n *NSQAdmin) handleAdminActions() {
	for action := range n.notifications {
		content, err := json.Marshal(action)
		if err != nil {
			n.logf(LOG_ERROR, "failed to serialize admin action - %s", err)
		}
		httpclient := &http.Client{
			Transport: http_api.NewDeadlineTransport(n.getOpts().HTTPClientConnectTimeout, n.getOpts().HTTPClientRequestTimeout),
		}
		n.logf(LOG_INFO, "POSTing notification to %s", n.getOpts().NotificationHTTPEndpoint)
		resp, err := httpclient.Post(n.getOpts().NotificationHTTPEndpoint,
			"application/json", bytes.NewBuffer(content))
		if err != nil {
Repository: nsqio/nsq
File: nsqadmin/static/js/app_state.js (JavaScript)

Code before:
            'STATSD_PREFIX': STATSD_PREFIX,
            'NSQLOOKUPD': NSQLOOKUPD,
            'graph_interval': '2h',
            'IS_ADMIN': IS_ADMIN,
            'BASE_PATH': BASE_PATH
        };
    },

    initialize: function() {
        this.on('change:graph_interval', function(model, v) {
            localStorage.setItem('graph_interval', v);
        });

        var qp = _.object(_.compact(_.map(window.location.search.slice(1).split('&'),
            function(item) { if (item) { return item.split('='); } })));

        var def = this.get('GRAPH_ENABLED') ? '2h' : 'off';
        var interval = qp['t'] || localStorage.getItem('graph_interval') || def;
        this.set('graph_interval', interval);
    },

    basePath: function(p) {
        // if base path is / then don't prefix
        var bp = this.get('BASE_PATH') == '/' ? '' : this.get('BASE_PATH');
        // trim trailing /
        return (bp + p).replace(/\/$/, '');
    },

    apiPath: function(p) {
        return this.basePath('/api' + p);
    }
});

var appState = new AppState();

window.AppState = appState;

module.exports = appState;


Review comment (bug):
Need to do a little testing but I _think_ there's a bug here when base path is `/` and this is passed a path `/`

Code after:
            'STATSD_PREFIX': STATSD_PREFIX,
            'NSQLOOKUPD': NSQLOOKUPD,
            'graph_interval': '2h',
            'IS_ADMIN': IS_ADMIN,
            'BASE_PATH': BASE_PATH
        };
    },

    initialize: function() {
        this.on('change:graph_interval', function(model, v) {
            localStorage.setItem('graph_interval', v);
        });

        var qp = _.object(_.compact(_.map(window.location.search.slice(1).split('&'),
            function(item) { if (item) { return item.split('='); } })));

        var def = this.get('GRAPH_ENABLED') ? '2h' : 'off';
        var interval = qp['t'] || localStorage.getItem('graph_interval') || def;
        this.set('graph_interval', interval);
    },

    basePath: function(p) {
        // if base path is / then don't prefix
        var bp = this.get('BASE_PATH') == '/' ? '' : this.get('BASE_PATH');
        // remove trailing /, but guarantee at least /
        return (bp + p).replace(/\/$/, '') || '/';
    },

    apiPath: function(p) {
        return this.basePath('/api' + p);
    }
});

var appState = new AppState();

window.AppState = appState;

module.exports = appState;
Repository: nsqio/nsq
File: nsqadmin/nsqadmin.go (Go)

Code before:
	for _, address := range opts.NSQLookupdHTTPAddresses {
		verifyAddress("--lookupd-http-address", address)
	}

	for _, address := range opts.NSQDHTTPAddresses {
		verifyAddress("--nsqd-http-address", address)
	}

	if opts.ProxyGraphite {
		url, err := url.Parse(opts.GraphiteURL)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --graphite-url='%s' - %s", opts.GraphiteURL, err)
			os.Exit(1)
		}
		n.graphiteURL = url
	}

	if opts.AllowConfigFromCIDR != "" {
		_, _, err := net.ParseCIDR(opts.AllowConfigFromCIDR)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --allow-config-from-cidr='%s' - %s", opts.AllowConfigFromCIDR, err)
			os.Exit(1)
		}
	}

	opts.BasePath = normalizeBasePath(path.Clean(opts.BasePath))

	n.logf(LOG_INFO, version.String("nsqadmin"))

	return n
}

func normalizeBasePath(p string) string {
	if len(p) == 0 {
		return "/"
	}
	// add leading slash
	if p[0] != '/' {
		p = "/" + p
	}
	return p
}

func (n *NSQAdmin) getOpts() *Options {
	return n.opts.Load().(*Options)
}

func (n *NSQAdmin) swapOpts(opts *Options) {
	n.opts.Store(opts)
}


Review comment (suggestion):
I didn't initially realize that `path.Clean()` removed the trailing slash. It's part of the normalization logic so it should probably go in `normalizeBasePath()`.

Code after:
	for _, address := range opts.NSQLookupdHTTPAddresses {
		verifyAddress("--lookupd-http-address", address)
	}

	for _, address := range opts.NSQDHTTPAddresses {
		verifyAddress("--nsqd-http-address", address)
	}

	if opts.ProxyGraphite {
		url, err := url.Parse(opts.GraphiteURL)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --graphite-url='%s' - %s", opts.GraphiteURL, err)
			os.Exit(1)
		}
		n.graphiteURL = url
	}

	if opts.AllowConfigFromCIDR != "" {
		_, _, err := net.ParseCIDR(opts.AllowConfigFromCIDR)
		if err != nil {
			n.logf(LOG_FATAL, "failed to parse --allow-config-from-cidr='%s' - %s", opts.AllowConfigFromCIDR, err)
			os.Exit(1)
		}
	}

	opts.BasePath = normalizeBasePath(opts.BasePath)

	n.logf(LOG_INFO, version.String("nsqadmin"))

	return n
}

func normalizeBasePath(p string) string {
	if len(p) == 0 {
		return "/"
	}
	// add leading slash
	if p[0] != '/' {
		p = "/" + p
	}
	return path.Clean(p)
}

func (n *NSQAdmin) getOpts() *Options {
	return n.opts.Load().(*Options)
}

func (n *NSQAdmin) swapOpts(opts *Options) {
	n.opts.Store(opts)
}
Repository: nsqio/nsq
File: nsqd/nsqd.go (Go)

Code before:
	TLSRequired
)

type errStore struct {
	err error
}

type Producer interface {
	Stats() ClientStats
}

type NSQD struct {
	// 64bit atomic vars need to be first for proper alignment on 32bit platforms
	clientIDSequence int64

	sync.RWMutex

	opts atomic.Value

	dl        *dirlock.DirLock
	isLoading int32
	errValue  atomic.Value
	startTime time.Time

	topicMap  map[string]*Topic
	producers map[int64]Producer

	lookupPeers atomic.Value

	tcpListener   net.Listener
	httpListener  net.Listener
	httpsListener net.Listener
	tlsConfig     *tls.Config

	poolSize int

	notifyChan           chan interface{}
	optsNotificationChan chan struct{}
	exitChan             chan int
	waitGroup            util.WaitGroupWrapper

	ci *clusterinfo.ClusterInfo
}

func New(opts *Options) *NSQD {
	dataPath := opts.DataPath
	if opts.DataPath == "" {
		cwd, _ := os.Getwd()
		dataPath = cwd
	}
	if opts.Logger == nil {

Review comment (suggestion):
Calling this variable `producers` seems misleading because we can't really know or categorize for sure. I'd rather call this `clients`.

Code after:
	TLSRequiredExceptHTTP
	TLSRequired
)

type errStore struct {
	err error
}

type Client interface {
	Stats() ClientStats
	IsProducer() bool
}

type NSQD struct {
	// 64bit atomic vars need to be first for proper alignment on 32bit platforms
	clientIDSequence int64

	sync.RWMutex

	opts atomic.Value

	dl        *dirlock.DirLock
	isLoading int32
	errValue  atomic.Value
	startTime time.Time

	topicMap map[string]*Topic

	clientLock sync.RWMutex
	clients    map[int64]Client

	lookupPeers atomic.Value

	tcpListener   net.Listener
	httpListener  net.Listener
	httpsListener net.Listener
	tlsConfig     *tls.Config

	poolSize int

	notifyChan           chan interface{}
	optsNotificationChan chan struct{}
	exitChan             chan int
	waitGroup            util.WaitGroupWrapper

	ci *clusterinfo.ClusterInfo
}

func New(opts *Options) *NSQD {
	dataPath := opts.DataPath
	if opts.DataPath == "" {