<commit_before>package native

import (
	"encoding/base64"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"sync"
	"time"

	"gopkg.in/inconshreveable/log15.v2"

	"github.com/CyCoreSystems/ari"

	"golang.org/x/net/context"
	"golang.org/x/net/websocket"
)

// Conn is a connection to a native ARI server
type Conn struct {
	Options Options // client options

	WSConfig *websocket.Config // websocket connection configuration

	ReadyChan chan struct{}

	Bus    ari.Bus        // event bus
	events chan ari.Event // chan on which events are sent

	httpClient http.Client

	ctx    context.Context
	cancel context.CancelFunc
	mu     sync.Mutex
}

func newConn(opts Options) (c *Conn) {

	if opts.Context == nil {
		opts.Context = context.Background()
	}

	c = &Conn{}
	c.Options = opts
	c.ctx, c.cancel = context.WithCancel(opts.Context)

	return
}

// Close closes the ARI client
func (c *Conn) Close() error {
	if c.cancel != nil {
		c.cancel()
	}
	return nil
}

// Listen maintains and listens to a websocket connection until told to stop.
func (c *Conn) Listen() (err error) {

	// Construct the websocket config, if we don't already have one
	if c.WSConfig == nil {
		// Construct the websocket connection url
		v := url.Values{}
		v.Set("app", c.Options.Application)
		wsurl := c.Options.WebsocketURL + "?" + v.Encode()

		// Construct a websocket.Config
		c.WSConfig, err = websocket.NewConfig(wsurl, "http://localhost/")
		if err != nil {
			Logger.Error("Failed to construct a valid websocket config:", err.Error())
			return fmt.Errorf("Failed to construct websocket config: %s", err.Error())
		}

		// Add the authorization header
		if c.Options.Username != "" && c.Options.Password != "" {
			c.WSConfig.Header.Set("Authorization", "Basic "+basicAuth(c.Options.Username, c.Options.Password))
		} else if os.Getenv("ARI_USERNAME") != "" {
			c.WSConfig.Header.Set("Authorization", "Basic "+basicAuth(os.Getenv("ARI_USERNAME"), os.Getenv("ARI_PASSWORD")))
		} else {
			Logger.Warn("No credentials found; expect failure")
		}
	}

	// Make sure the bus is set up
	if c.Bus == nil {
		c.Bus = startbus(c.ctx)
	}

	// Make sure we have a readychan to signal the websocket is up
	if c.ReadyChan == nil {
		c.ReadyChan = make(chan struct{})
	}

	// If we are in test mode, do not connect the websocket, but
	// return and close the ready channel.
	//if c.TestMode {
	//	close(c.ReadyChan)
	//	return nil
	//}

	// Setup and listen on the websocket
	go c.listen(c.ctx)

	// Wait for the websocket connection to connect or for the context to be cancelled
	select {
	case <-c.ReadyChan:
	case <-c.ctx.Done():
		return c.ctx.Err()
	}

	return nil
}

func (c *Conn) listen(ctx context.Context) {
	var ws *websocket.Conn
	var err error

	// Close the websocket if our context is closed
	go func() {
		<-ctx.Done()
		Logger.Info("closing websocket on request")
		if ws != nil {
			ws.Close()
		}
	}()

	for {

		select {
		case <-ctx.Done():
			return
		default:
		}

		Logger.Debug("Connecting to websocket")

		ws, err = websocket.DialConfig(c.WSConfig)
		if err != nil {
			Logger.Error("Failed to create websocket connection to Asterisk", "error", err.Error())
			time.Sleep(1 * time.Second)
		}

		close(c.ReadyChan)

		err = c.wsRead(ws)
		if err != nil {
			Logger.Error("Failure reading from websocket", "error", err.Error())
		}

		// Clean up
		ws.Close()
		ws = nil

		c.ReadyChan = make(chan struct{})
<commit_msg>v3 - native client - fix crash when websocket connection isn't ready yet<commit_after>package native

import (
	"encoding/base64"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"sync"
	"time"

	"gopkg.in/inconshreveable/log15.v2"

	"github.com/CyCoreSystems/ari"

	"golang.org/x/net/context"
	"golang.org/x/net/websocket"
)

// Conn is a connection to a native ARI server
type Conn struct {
	Options Options // client options

	WSConfig *websocket.Config // websocket connection configuration

	ReadyChan chan struct{}

	Bus    ari.Bus        // event bus
	events chan ari.Event // chan on which events are sent

	httpClient http.Client

	ctx    context.Context
	cancel context.CancelFunc
	mu     sync.Mutex
}

func newConn(opts Options) (c *Conn) {

	if opts.Context == nil {
		opts.Context = context.Background()
	}

	c = &Conn{}
	c.Options = opts
	c.ctx, c.cancel = context.WithCancel(opts.Context)

	return
}

// Close closes the ARI client
func (c *Conn) Close() error {
	if c.cancel != nil {
		c.cancel()
	}
	return nil
}

// Listen maintains and listens to a websocket connection until told to stop.
func (c *Conn) Listen() (err error) {

	// Construct the websocket config, if we don't already have one
	if c.WSConfig == nil {
		// Construct the websocket connection url
		v := url.Values{}
		v.Set("app", c.Options.Application)
		wsurl := c.Options.WebsocketURL + "?" + v.Encode()

		// Construct a websocket.Config
		c.WSConfig, err = websocket.NewConfig(wsurl, "http://localhost/")
		if err != nil {
			Logger.Error("Failed to construct a valid websocket config:", err.Error())
			return fmt.Errorf("Failed to construct websocket config: %s", err.Error())
		}

		// Add the authorization header
		if c.Options.Username != "" && c.Options.Password != "" {
			c.WSConfig.Header.Set("Authorization", "Basic "+basicAuth(c.Options.Username, c.Options.Password))
		} else if os.Getenv("ARI_USERNAME") != "" {
			c.WSConfig.Header.Set("Authorization", "Basic "+basicAuth(os.Getenv("ARI_USERNAME"), os.Getenv("ARI_PASSWORD")))
		} else {
			Logger.Warn("No credentials found; expect failure")
		}
	}

	// Make sure the bus is set up
	if c.Bus == nil {
		c.Bus = startbus(c.ctx)
	}

	// Make sure we have a readychan to signal the websocket is up
	if c.ReadyChan == nil {
		c.ReadyChan = make(chan struct{})
	}

	// If we are in test mode, do not connect the websocket, but
	// return and close the ready channel.
	//if c.TestMode {
	//	close(c.ReadyChan)
	//	return nil
	//}

	// Setup and listen on the websocket
	go c.listen(c.ctx)

	// Wait for the websocket connection to connect or for the context to be cancelled
	select {
	case <-c.ReadyChan:
	case <-c.ctx.Done():
		return c.ctx.Err()
	}

	return nil
}

func (c *Conn) listen(ctx context.Context) {
	var ws *websocket.Conn
	var err error

	// Close the websocket if our context is closed
	go func() {
		<-ctx.Done()
		Logger.Info("closing websocket on request")
		if ws != nil {
			ws.Close()
		}
	}()

	for {

		select {
		case <-ctx.Done():
			return
		default:
		}

		Logger.Debug("Connecting to websocket")

		ws, err = websocket.DialConfig(c.WSConfig)
		if err != nil {
			Logger.Error("Failed to create websocket connection to Asterisk", "error", err.Error())
			time.Sleep(1 * time.Second)
			continue
		}

		close(c.ReadyChan)

		err = c.wsRead(ws)
		if err != nil {
			Logger.Error("Failure reading from websocket", "error", err.Error())
		}

		// Clean up
		ws.Close()
		ws = nil

		c.ReadyChan = make(chan struct{})

{'fraction_non_alphanumeric': 0.1355119825708061, 'fraction_numerical': 0.001888162672476398, 'mean_word_length': 3.1708055723803756, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, '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': '1509930', 'n_tokens_mistral': 2582, 'n_tokens_neox': 2336, 'n_words': 703}