<commit_before>package fh

import (
	"errors"
	"log"
	"net"
	"sync"
	"time"

	"github.com/gabrielperezs/firehose-pool"
	"github.com/gallir/radix.improved/redis"
	"github.com/gallir/smart-relayer/lib"
)

// Server is the thread that listen for clients' connections
type Server struct {
	sync.Mutex
	config   lib.RelayerConfig
	done     chan bool
	exiting  bool
	reseting bool
	listener net.Listener

	fh             *firehosePool.Server
	lastConnection time.Time
	lastError      time.Time
	errors         int64
}

const (
	maxConnections      = 2
	requestBufferSize   = 1024 * 10
	maxConnectionsTries = 3
	connectionRetry     = 5 * time.Second
	errorsFrame         = 10 * time.Second
	maxErrors           = 10 // Limit of errors to restart the connection
	connectTimeout      = 15 * time.Second
)

var (
	errBadCmd      = errors.New("ERR bad command")
	errKO          = errors.New("fatal error")
	errOverloaded  = errors.New("Redis overloaded")
	respOK         = redis.NewRespSimple("OK")
	respTrue       = redis.NewResp(1)
	respBadCommand = redis.NewResp(errBadCmd)
	respKO         = redis.NewResp(errKO)
	commands       map[string]*redis.Resp
)

func init() {
	commands = map[string]*redis.Resp{
		"PING":   respOK,
		"MULTI":  respOK,
		"EXEC":   respOK,
		"SET":    respOK,
		"SADD":   respOK,
		"HMSET":  respOK,
		"CSET":   respOK,
		"CSADD":  respOK,
		"CHMSET": respOK,
		"RAWSET": respOK,
	}
}

// New creates a new Redis local server
func New(c lib.RelayerConfig, done chan bool) (*Server, error) {
	srv := &Server{
		done:   done,
		errors: 0,
	}

	srv.Reload(&c)

	return srv, nil
}

// Reload the configuration
func (srv *Server) Reload(c *lib.RelayerConfig) (err error) {
	srv.Lock()
	defer srv.Unlock()

	srv.config = *c

	if srv.config.MaxConnections <= 0 {
		srv.config.MaxConnections = maxConnections
	}

	if srv.config.Buffer == 0 {
		srv.config.Buffer = requestBufferSize
	}

	fhConfig := firehosePool.Config{
		Profile:       srv.config.Profile,
		Region:        srv.config.Region,
		StreamName:    srv.config.StreamName,
		MaxWorkers:    srv.config.MaxConnections,
		MaxRecords:    srv.config.MaxRecords,
		Buffer:        srv.config.Buffer,
		ConcatRecords: srv.config.Concat,
	}

	if srv.fh == nil {
		srv.fh = firehosePool.New(fhConfig)
	} else {
		go srv.fh.Reload(&fhConfig)
	}

	return nil
}

// Start accepts incoming connections on the Listener
func (srv *Server) Start() (e error) {
	srv.Lock()
	defer srv.Unlock()

	srv.listener, e = lib.NewListener(srv.config)
	if e != nil {
		return e
	}

	// Serve clients
	go func(l net.Listener) {
		defer srv.listener.Close()
		for {
			netConn, e := l.Accept()
			if e != nil {
				if netErr, ok := e.(net.Error); ok && netErr.Timeout() {
					// Paranoid, ignore timeout errors
					log.Println("Firehose ERROR: timeout at local listener", srv.config.ListenHost(), e)
					continue
				}
				if srv.exiting {
					log.Println("Firehose: exiting local listener", srv.config.ListenHost())
					return
				}
				log.Fatalln("Firehose ERROR: emergency error in local listener", srv.config.ListenHost(), e)
				return
			}
			go srv.handleConnection(netConn)
		}
	}(srv.listener)

	return
}

// Exit closes the listener and send done to main
func (srv *Server) Exit() {
	srv.exiting = true

	if srv.listener != nil {
		srv.listener.Close()
	}

	go srv.fh.Exit()

	srv.fh.Waiting()

	// finishing the server
	srv.done <- true
}

func (srv *Server) sendRecord(r *lib.InterRecord) {
	if srv.reseting || srv.exiting {
		return
	}

	go func() {
		select {
		case srv.fh.C <- r.Bytes():
		default:
			lib.Debugf("Firehose: channel is full. Queued messages %d", len(srv.fh.C))
		}
	}()
}

func (srv *Server) sendBytes(b []byte) {
	r := lib.NewInterRecord()
	r.Types = 1
	r.Raw = b
	srv.sendRecord(r)
}
<commit_msg>Removed useles variable and temporary to github.com/gallir/firehose-pool<commit_after>package fh

import (
	"errors"
	"log"
	"net"
	"sync"
	"time"

	"github.com/gallir/firehose-pool"
	"github.com/gallir/radix.improved/redis"
	"github.com/gallir/smart-relayer/lib"
)

// Server is the thread that listen for clients' connections
type Server struct {
	sync.Mutex
	config   lib.RelayerConfig
	done     chan bool
	exiting  bool
	listener net.Listener

	fh             *firehosePool.Server
	lastConnection time.Time
	lastError      time.Time
	errors         int64
}

const (
	maxConnections      = 2
	requestBufferSize   = 1024 * 10
	maxConnectionsTries = 3
	connectionRetry     = 5 * time.Second
	errorsFrame         = 10 * time.Second
	maxErrors           = 10 // Limit of errors to restart the connection
	connectTimeout      = 15 * time.Second
)

var (
	errBadCmd      = errors.New("ERR bad command")
	errKO          = errors.New("fatal error")
	errOverloaded  = errors.New("Redis overloaded")
	respOK         = redis.NewRespSimple("OK")
	respTrue       = redis.NewResp(1)
	respBadCommand = redis.NewResp(errBadCmd)
	respKO         = redis.NewResp(errKO)
	commands       map[string]*redis.Resp
)

func init() {
	commands = map[string]*redis.Resp{
		"PING":   respOK,
		"MULTI":  respOK,
		"EXEC":   respOK,
		"SET":    respOK,
		"SADD":   respOK,
		"HMSET":  respOK,
		"CSET":   respOK,
		"CSADD":  respOK,
		"CHMSET": respOK,
		"RAWSET": respOK,
	}
}

// New creates a new Redis local server
func New(c lib.RelayerConfig, done chan bool) (*Server, error) {
	srv := &Server{
		done:   done,
		errors: 0,
	}

	srv.Reload(&c)

	return srv, nil
}

// Reload the configuration
func (srv *Server) Reload(c *lib.RelayerConfig) (err error) {
	srv.Lock()
	defer srv.Unlock()

	srv.config = *c

	if srv.config.MaxConnections <= 0 {
		srv.config.MaxConnections = maxConnections
	}

	if srv.config.Buffer == 0 {
		srv.config.Buffer = requestBufferSize
	}

	fhConfig := firehosePool.Config{
		Profile:       srv.config.Profile,
		Region:        srv.config.Region,
		StreamName:    srv.config.StreamName,
		MaxWorkers:    srv.config.MaxConnections,
		MaxRecords:    srv.config.MaxRecords,
		Buffer:        srv.config.Buffer,
		ConcatRecords: srv.config.Concat,
	}

	if srv.fh == nil {
		srv.fh = firehosePool.New(fhConfig)
	} else {
		go srv.fh.Reload(&fhConfig)
	}

	return nil
}

// Start accepts incoming connections on the Listener
func (srv *Server) Start() (e error) {
	srv.Lock()
	defer srv.Unlock()

	srv.listener, e = lib.NewListener(srv.config)
	if e != nil {
		return e
	}

	// Serve clients
	go func(l net.Listener) {
		defer srv.listener.Close()
		for {
			netConn, e := l.Accept()
			if e != nil {
				if netErr, ok := e.(net.Error); ok && netErr.Timeout() {
					// Paranoid, ignore timeout errors
					log.Println("Firehose ERROR: timeout at local listener", srv.config.ListenHost(), e)
					continue
				}
				if srv.exiting {
					log.Println("Firehose: exiting local listener", srv.config.ListenHost())
					return
				}
				log.Fatalln("Firehose ERROR: emergency error in local listener", srv.config.ListenHost(), e)
				return
			}
			go srv.handleConnection(netConn)
		}
	}(srv.listener)

	return
}

// Exit closes the listener and send done to main
func (srv *Server) Exit() {
	srv.exiting = true

	if srv.listener != nil {
		srv.listener.Close()
	}

	go srv.fh.Exit()

	srv.fh.Waiting()

	// finishing the server
	srv.done <- true
}

func (srv *Server) sendRecord(r *lib.InterRecord) {
	if srv.exiting {
		return
	}

	select {
	case srv.fh.C <- r.Bytes():
	default:
		lib.Debugf("Firehose: channel is full. Queued messages %d", len(srv.fh.C))
	}
}

func (srv *Server) sendBytes(b []byte) {
	r := lib.NewInterRecord()
	r.Types = 1
	r.Raw = b
	srv.sendRecord(r)
}

{'fraction_non_alphanumeric': 0.1281473628412404, 'fraction_numerical': 0.0058309037900874635, 'mean_word_length': 2.9554507337526204, 'pattern_counts': {'":': 20, '<': 9, '<?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': '6491336', 'n_tokens_mistral': 3109, 'n_tokens_neox': 2804, 'n_words': 665}