<commit_before>// Copyright 2015 Apcera Inc. All rights reserved.

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"math/rand"

	"github.com/nats-io/nats"
)

// TODO:  Create interfaces and reuse code w/ stan-client-sim

// Some sane defaults
const (
	DefaultConnectWait    = 4 * time.Minute
	DefaultConfigFileName = "config.json"
	UniqueSubject         = "UNIQUE"
)

var trace bool
var verbose bool

func verbosef(format string, v ...interface{}) {
	if verbose {
		log.Printf(format, v...)
	}
}

// ClientSubConfig represents a subscription for a client
type ClientSubConfig struct {
	Count   int    `json:"count"`
	Subject string `json:"subject"`
}

// ClientConfig represents a streaming client
type ClientConfig struct {
	Name           string            `json:"name"`
	Instances      int               `json:"instances"`
	UserName       string            `json:"username"`
	Password       string            `json:"password"`
	PubMsgSize     int               `json:"pub_msgsize"`
	PubRate        string            `json:"pub_delay"`
	PubMsgCount    int               `json:"pub_msgcount"`
	PublishSubject string            `json:"pub_subject"`
	Subscriptions  []ClientSubConfig `json:"subscriptions"`
}

// Config is the server configuration
type Config struct {
	MaxStartDelay int            `json:"client_start_delay_max"`
	ServerURLs    string         `json:"url"`
	TLSClientCA   string         `json:"tlsca"`
	TLSClientCert string         `json:"tlscert"`
	TLSClientKey  string         `json:"tlskey"`
	UseTLS        bool           `json:"usetls"`
	Clients       []ClientConfig `json:"clients"`
}

// ClientSub is a client subscription
type ClientSub struct {
	subject  string
	sub      *nats.Subscription
	ch       chan (bool)
	received int32
	max      int32
	isDone   bool
}

// GetReceivedCount returns the count of received messages
func (cs *ClientSub) GetReceivedCount() int32 {
	return atomic.LoadInt32(&cs.received)
}

// Client represents a NATS streaming client
type Client struct {
	sync.Mutex
	cm                *ClientManager
	config            *ClientConfig
	clientID          string
	hasUniqueSubjects bool
	nc                *nats.Conn
	subs              []*ClientSub
	publishCount      int32
	publishDelay      time.Duration
	subCh             chan (bool)
	payload           []byte
	done              bool
}

// NewClient returns a new client.
func NewClient(config *ClientConfig, instance int, cm *ClientManager) *Client {
	c := &Client{}
	c.cm = cm
	c.config = config
	c.clientID = fmt.Sprintf("%s-%d", c.config.Name, instance)

	if c.isPublisher() {
		c.publishDelay = parsePubRate(c.config.PubRate)
	}
	return c
}

func (c *Client) connect() error {
	var err error
	opts := nats.DefaultOptions
	opts.Servers = strings.Split(c.cm.config.ServerURLs, ",")
	for i, s := range opts.Servers {
		opts.Servers[i] = strings.Trim(s, " ")
	}

	opts.Secure = c.cm.config.UseTLS
	opts.AsyncErrorCB = errorHandler
	opts.DisconnectedCB = disconnectedHandler
	opts.ReconnectedCB = reconnectedHandler
	opts.ClosedCB = closedHandler
	opts.User = c.config.UserName
	opts.Password = <PASSWORD>
	opts.Name = c.clientID
	c.nc, err = opts.Connect()<commit_msg>attempt initial connection mulitiple times<commit_after>// Copyright 2015 Apcera Inc. All rights reserved.

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"math/rand"

	"github.com/nats-io/nats"
)

// TODO:  Create interfaces and reuse code w/ stan-client-sim

// Some sane defaults
const (
	DefaultConnectWait    = 4 * time.Minute
	DefaultConfigFileName = "config.json"
	UniqueSubject         = "UNIQUE"
)

var trace bool
var verbose bool

func verbosef(format string, v ...interface{}) {
	if verbose {
		log.Printf(format, v...)
	}
}

// ClientSubConfig represents a subscription for a client
type ClientSubConfig struct {
	Count   int    `json:"count"`
	Subject string `json:"subject"`
}

// ClientConfig represents a streaming client
type ClientConfig struct {
	Name           string            `json:"name"`
	Instances      int               `json:"instances"`
	UserName       string            `json:"username"`
	Password       string            `json:"password"`
	PubMsgSize     int               `json:"pub_msgsize"`
	PubRate        string            `json:"pub_delay"`
	PubMsgCount    int               `json:"pub_msgcount"`
	PublishSubject string            `json:"pub_subject"`
	Subscriptions  []ClientSubConfig `json:"subscriptions"`
}

// Config is the server configuration
type Config struct {
	MaxStartDelay int            `json:"client_start_delay_max"`
	ServerURLs    string         `json:"url"`
	TLSClientCA   string         `json:"tlsca"`
	TLSClientCert string         `json:"tlscert"`
	TLSClientKey  string         `json:"tlskey"`
	UseTLS        bool           `json:"usetls"`
	Clients       []ClientConfig `json:"clients"`
}

// ClientSub is a client subscription
type ClientSub struct {
	subject  string
	sub      *nats.Subscription
	ch       chan (bool)
	received int32
	max      int32
	isDone   bool
}

// GetReceivedCount returns the count of received messages
func (cs *ClientSub) GetReceivedCount() int32 {
	return atomic.LoadInt32(&cs.received)
}

// Client represents a NATS streaming client
type Client struct {
	sync.Mutex
	cm                *ClientManager
	config            *ClientConfig
	clientID          string
	hasUniqueSubjects bool
	nc                *nats.Conn
	subs              []*ClientSub
	publishCount      int32
	publishDelay      time.Duration
	subCh             chan (bool)
	payload           []byte
	done              bool
}

// NewClient returns a new client.
func NewClient(config *ClientConfig, instance int, cm *ClientManager) *Client {
	c := &Client{}
	c.cm = cm
	c.config = config
	c.clientID = fmt.Sprintf("%s-%d", c.config.Name, instance)

	if c.isPublisher() {
		c.publishDelay = parsePubRate(c.config.PubRate)
	}
	return c
}

func (c *Client) connect() error {
	var err error
	opts := nats.DefaultOptions
	opts.Servers = strings.Split(c.cm.config.ServerURLs, ",")
	for i, s := range opts.Servers {
		opts.Servers[i] = strings.Trim(s, " ")
	}

	opts.Secure = c.cm.config.UseTLS
	opts.AsyncErrorCB = errorHandler
	opts.DisconnectedCB = disconnectedHandler
	opts.ReconnectedCB = reconnectedHandler
	opts.ClosedCB = closedHandler
	opts.User = c.config.UserName
	opts.Password = c.config.Password
	opts.Name = c.clientID
	c.nc, err = opts.Connect()

	// if we can't connect via error, keep trying - this is
	// a stress test.
	if err != nats.ErrNoServers {
		for i := 0; i < 10; i++ {
			log.Printf("%s:  retrying initial connect.  %v\n", c.clientID, err)
			time.Sleep(100 * time.Millisecond)
			c.nc, err = opts.Connect()
		}
	}
{'fraction_non_alphanumeric': 0.10608286805759624, 'fraction_numerical': 0.005289450484866294, 'mean_word_length': 2.719672131147541, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2251195', 'n_tokens_mistral': 2213, 'n_tokens_neox': 2138, 'n_words': 581}