<commit_before>package consul

import (
	"fmt"
	"net"
	"strings"
	"time"

	"github.com/gorilla/websocket"
	api "github.com/hashicorp/consul/api"
	observer "github.com/imkira/go-observer"
	"github.com/jippi/hashi-ui/backend/structs"
	"github.com/jippi/hashi-ui/backend/subscriber"
	uuid "github.com/satori/go.uuid"
)

// Connection monitors the websocket connection. It processes any action
// received on the websocket and sends out actions on Consul state changes. It
// maintains a set to keep track of the running watches.
type Connection struct {
	ID                uuid.UUID
	shortID           string
	socket            *websocket.Conn
	receive           chan *structs.Action
	send              chan *structs.Action
	destroyCh         chan struct{}
	watches           *subscriber.Manager
	hub               *Hub
	region            *Region
	broadcastChannels *RegionBroadcastChannels
}

// NewConnection creates a new connection.
func NewConnection(hub *Hub, socket *websocket.Conn, consulRegion *Region, channels *RegionBroadcastChannels) *Connection {
	connectionID := uuid.NewV4()

	return &Connection{
		ID:                connectionID,
		shortID:           fmt.Sprintf("%s", connectionID)[0:8],
		watches:           &subscriber.Manager{},
		hub:               hub,
		socket:            socket,
		receive:           make(chan *structs.Action),
		send:              make(chan *structs.Action),
		destroyCh:         make(chan struct{}),
		region:            consulRegion,
		broadcastChannels: channels,
	}
}

// Warningf is a stupid wrapper for logger.Warningf
func (c *Connection) Warningf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Warningf(message, args...)
}

// Errorf is a stupid wrapper for logger.Errorf
func (c *Connection) Errorf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Errorf(message, args...)
}

// Infof is a stupid wrapper for logger.Infof
func (c *Connection) Infof(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Infof(message, args...)
}

// Debugf is a stupid wrapper for logger.Debugf
func (c *Connection) Debugf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Debugf(message, args...)
}

func (c *Connection) writePump() {
	defer c.socket.Close()

	for {
		select {
		case <-c.destroyCh:
			c.Warningf("Stopping writePump")
			return

		case action, ok := <-c.send:
			if !ok {
				if err := c.socket.WriteMessage(websocket.CloseMessage, []byte{}); err != nil {
					c.Errorf("Could not write close message to websocket: %s", err)
				}
				return
			}

			if err := c.socket.WriteJSON(action); err != nil {
				c.Errorf("Could not write action to websocket: %s", err)
			}
		}
	}
}

func (c *Connection) readPump() {
	var action structs.Action

	for {
		err := c.socket.ReadJSON(&action)
		if err != nil {
			break
		}
<commit_msg>also close consul writepump on error<commit_after>package consul

import (
	"fmt"
	"net"
	"strings"
	"time"

	"github.com/gorilla/websocket"
	api "github.com/hashicorp/consul/api"
	observer "github.com/imkira/go-observer"
	"github.com/jippi/hashi-ui/backend/structs"
	"github.com/jippi/hashi-ui/backend/subscriber"
	uuid "github.com/satori/go.uuid"
)

// Connection monitors the websocket connection. It processes any action
// received on the websocket and sends out actions on Consul state changes. It
// maintains a set to keep track of the running watches.
type Connection struct {
	ID                uuid.UUID
	shortID           string
	socket            *websocket.Conn
	receive           chan *structs.Action
	send              chan *structs.Action
	destroyCh         chan struct{}
	watches           *subscriber.Manager
	hub               *Hub
	region            *Region
	broadcastChannels *RegionBroadcastChannels
}

// NewConnection creates a new connection.
func NewConnection(hub *Hub, socket *websocket.Conn, consulRegion *Region, channels *RegionBroadcastChannels) *Connection {
	connectionID := uuid.NewV4()

	return &Connection{
		ID:                connectionID,
		shortID:           fmt.Sprintf("%s", connectionID)[0:8],
		watches:           &subscriber.Manager{},
		hub:               hub,
		socket:            socket,
		receive:           make(chan *structs.Action),
		send:              make(chan *structs.Action),
		destroyCh:         make(chan struct{}),
		region:            consulRegion,
		broadcastChannels: channels,
	}
}

// Warningf is a stupid wrapper for logger.Warningf
func (c *Connection) Warningf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Warningf(message, args...)
}

// Errorf is a stupid wrapper for logger.Errorf
func (c *Connection) Errorf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Errorf(message, args...)
}

// Infof is a stupid wrapper for logger.Infof
func (c *Connection) Infof(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Infof(message, args...)
}

// Debugf is a stupid wrapper for logger.Debugf
func (c *Connection) Debugf(format string, args ...interface{}) {
	message := fmt.Sprintf("[%s] ", c.shortID) + format
	logger.Debugf(message, args...)
}

func (c *Connection) writePump() {
	defer c.socket.Close()

	for {
		select {
		case <-c.destroyCh:
			c.Warningf("Stopping writePump")
			return

		case action, ok := <-c.send:
			if !ok {
				if err := c.socket.WriteMessage(websocket.CloseMessage, []byte{}); err != nil {
					c.Errorf("Could not write close message to websocket: %s", err)
				}
				return
			}

			if err := c.socket.WriteJSON(action); err != nil {
				c.Errorf("Could not write action to websocket: %s", err)
				return
			}
		}
	}
}

func (c *Connection) readPump() {
	var action structs.Action

	for {
		err := c.socket.ReadJSON(&action)
		if err != nil {
			break
		}

{'fraction_non_alphanumeric': 0.12967914438502673, 'fraction_numerical': 0.001002673796791444, 'mean_word_length': 2.888888888888889, 'pattern_counts': {'":': 0, '<': 7, '<?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': '4249958', 'n_tokens_mistral': 2027, 'n_tokens_neox': 1889, 'n_words': 524}